概述
PromQL（Prometheus Query Language）是Prometheus的函数式查询语言，允许用户实时选择和聚合时间序列数据。PromQL专门为时间序列数据设计，提供了丰富的操作符和函数。
学习目标
通过本章学习，你将能够： - 掌握PromQL的基本语法和概念 - 学会使用各种选择器和操作符 - 理解聚合函数和时间函数的使用 - 编写复杂的查询表达式 - 优化查询性能
PromQL基础语法
1. 基本概念
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional, Any, Union
from datetime import datetime, timedelta
import re
class SelectorType(Enum):
"""选择器类型"""
INSTANT_VECTOR = "instant_vector"
RANGE_VECTOR = "range_vector"
SCALAR = "scalar"
STRING = "string"
class MatcherType(Enum):
"""匹配器类型"""
EQUAL = "="
NOT_EQUAL = "!="
REGEX_MATCH = "=~"
REGEX_NOT_MATCH = "!~"
class OperatorType(Enum):
"""操作符类型"""
# 算术操作符
ADD = "+"
SUBTRACT = "-"
MULTIPLY = "*"
DIVIDE = "/"
MODULO = "%"
POWER = "^"
# 比较操作符
EQUAL_COMP = "=="
NOT_EQUAL_COMP = "!="
GREATER = ">"
LESS = "<"
GREATER_EQUAL = ">="
LESS_EQUAL = "<="
# 逻辑操作符
AND = "and"
OR = "or"
UNLESS = "unless"
@dataclass
class LabelMatcher:
"""标签匹配器"""
label: str
operator: MatcherType
value: str
def __str__(self) -> str:
return f'{self.label}{self.operator.value}"{self.value}"'
@dataclass
class TimeRange:
"""时间范围"""
duration: str # 如: 5m, 1h, 1d
def to_seconds(self) -> int:
"""转换为秒数"""
unit_map = {
's': 1, 'm': 60, 'h': 3600, 'd': 86400, 'w': 604800, 'y': 31536000
}
match = re.match(r'(\d+)([smhdwy])', self.duration)
if match:
value, unit = match.groups()
return int(value) * unit_map[unit]
return 0
class PromQLBasics:
"""PromQL基础语法"""
def __init__(self):
self.examples = {}
def explain_data_types(self) -> str:
"""解释PromQL数据类型"""
return """
PromQL 数据类型：
1. 瞬时向量 (Instant Vector)
- 包含每个时间序列的单个样本
- 所有样本共享相同的时间戳
- 示例：http_requests_total
2. 范围向量 (Range Vector)
- 包含每个时间序列在时间范围内的样本集合
- 用于计算变化率等
- 示例：http_requests_total[5m]
3. 标量 (Scalar)
- 简单的数字浮点值
- 示例：2.5, 100
4. 字符串 (String)
- 简单的字符串值
- 目前使用有限
- 示例："hello"
数据类型转换：
- 瞬时向量 → 标量：聚合函数
- 范围向量 → 瞬时向量：rate(), increase()等函数
- 标量 → 瞬时向量：vector()函数
"""
def explain_selectors(self) -> str:
"""解释选择器语法"""
return """
选择器语法：
1. 基本选择器
http_requests_total # 选择指标名称
2. 标签匹配器
http_requests_total{job="api"} # 等于匹配
http_requests_total{status!="200"} # 不等于匹配
http_requests_total{method=~"GET|POST"} # 正则匹配
http_requests_total{path!~"/admin.*"} # 正则不匹配
3. 多标签匹配
http_requests_total{job="api", method="GET", status="200"}
4. 范围选择器
http_requests_total[5m] # 过去5分钟的数据
http_requests_total{job="api"}[1h] # 过去1小时的数据
5. 偏移修饰符
http_requests_total offset 1h # 1小时前的数据
http_requests_total[5m] offset 1d # 1天前的5分钟范围数据
6. @ 修饰符（时间点查询）
http_requests_total @ 1609459200 # 特定时间戳的数据
http_requests_total[5m] @ 1609459200 # 特定时间点前5分钟的数据
时间单位：
- s: 秒
- m: 分钟
- h: 小时
- d: 天
- w: 周
- y: 年
"""
def create_basic_examples(self) -> Dict[str, List[str]]:
"""创建基础查询示例"""
return {
"简单选择器": [
"up", # 所有up指标
"prometheus_build_info", # Prometheus构建信息
"node_cpu_seconds_total", # CPU时间总计
],
"标签过滤": [
'up{job="prometheus"}', # 特定job的up指标
'node_cpu_seconds_total{mode="idle"}', # idle模式的CPU时间
'http_requests_total{status!="200"}', # 非200状态的请求
],
"正则匹配": [
'up{job=~"prometheus|node"}', # job为prometheus或node
'node_filesystem_size_bytes{mountpoint=~"/.*"}', # 根目录挂载点
'http_requests_total{path!~"/health.*"}', # 非健康检查路径
],
"范围查询": [
"up[5m]", # 过去5分钟的up指标
"rate(http_requests_total[5m])", # 5分钟内的请求率
"increase(node_cpu_seconds_total[1h])", # 1小时内CPU时间增量
],
"时间偏移": [
"up offset 1h", # 1小时前的up指标
"rate(http_requests_total[5m] offset 1d)", # 1天前的请求率
"up @ 1609459200", # 特定时间戳的up指标
]
}
class PromQLOperators:
"""PromQL操作符"""
def explain_arithmetic_operators(self) -> str:
"""解释算术操作符"""
return """
算术操作符：
1. 基本算术操作
+ 加法
- 减法
* 乘法
/ 除法
% 取模
^ 幂运算
2. 向量与标量运算
node_memory_MemTotal_bytes / 1024 / 1024 # 转换为MB
cpu_usage_percent * 100 # 转换为百分比
disk_usage_bytes + 1000000 # 增加固定值
3. 向量与向量运算
# 相同标签的时间序列进行运算
node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes # 已用内存
# 一对多匹配
method_code:http_requests:rate5m / ignoring(code) group_left method:http_requests:rate5m
# 多对一匹配
instance_memory_limit_bytes / on(instance) group_right instance:node_memory_utilisation:ratio
4. 匹配修饰符
on(label1, label2) # 只基于指定标签匹配
ignoring(label1, label2) # 忽略指定标签进行匹配
group_left # 左侧向量的标签保留到结果中
group_right # 右侧向量的标签保留到结果中
示例：
# 计算内存使用率
(
node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes
) / node_memory_MemTotal_bytes * 100
# 计算磁盘使用率
(
node_filesystem_size_bytes - node_filesystem_free_bytes
) / node_filesystem_size_bytes * 100
"""
def explain_comparison_operators(self) -> str:
"""解释比较操作符"""
return """
比较操作符：
1. 基本比较操作
== 等于
!= 不等于
> 大于
< 小于
>= 大于等于
<= 小于等于
2. 过滤用法
# 过滤出CPU使用率大于80%的实例
cpu_usage_percent > 80
# 过滤出可用内存小于1GB的实例
node_memory_MemAvailable_bytes < 1024*1024*1024
# 过滤出磁盘使用率大于等于90%的挂载点
disk_usage_percent >= 90
3. 布尔修饰符
# 返回布尔值而不是过滤
cpu_usage_percent > bool 80 # 返回1或0
# 与聚合函数结合
count(cpu_usage_percent > 80) # 统计CPU使用率大于80%的实例数
4. 实际应用示例
# 高CPU使用率告警
avg(cpu_usage_percent) by (instance) > 85
# 低磁盘空间告警
(
node_filesystem_avail_bytes / node_filesystem_size_bytes * 100
) < 10
# 高内存使用率告警
(
1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)
) * 100 > 90
"""
def explain_logical_operators(self) -> str:
"""解释逻辑操作符"""
return """
逻辑操作符：
1. and（与）
# 两个条件都满足
cpu_usage_percent > 80 and memory_usage_percent > 90
# 交集操作
up{job="node"} and on(instance) cpu_usage_percent > 80
2. or（或）
# 任一条件满足
cpu_usage_percent > 90 or memory_usage_percent > 95
# 并集操作
up{job="prometheus"} or up{job="alertmanager"}
3. unless（除非）
# 左侧存在但右侧不存在
up unless on(instance) cpu_usage_percent > 80
# 排除特定条件
http_requests_total unless http_requests_total{path=~"/health.*"}
4. 复杂逻辑组合
# 高负载但非维护窗口
(
cpu_usage_percent > 80 or memory_usage_percent > 90
) unless on(instance) maintenance_mode == 1
# 服务异常但非已知问题
up == 0 unless on(instance) known_issues{status="active"}
5. 向量匹配
# 使用on()指定匹配标签
cpu_usage_percent > 80 and on(instance) memory_usage_percent > 90
# 使用ignoring()忽略特定标签
up and ignoring(job) cpu_usage_percent > 80
"""
class PromQLFunctions:
"""PromQL函数"""
def explain_rate_functions(self) -> str:
"""解释速率函数"""
return """
速率函数：
1. rate()
# 计算每秒平均增长率
rate(http_requests_total[5m]) # 5分钟内每秒请求数
rate(node_cpu_seconds_total[5m]) # 5分钟内每秒CPU时间
特点：
- 自动处理计数器重置
- 返回每秒的平均速率
- 适用于Counter类型指标
2. irate()
# 计算瞬时增长率（基于最后两个数据点）
irate(http_requests_total[5m]) # 瞬时请求率
特点：
- 更敏感，反应更快
- 基于最近两个样本
- 适合检测突发变化
3. increase()
# 计算时间范围内的增量
increase(http_requests_total[1h]) # 1小时内请求增量
特点：
- 返回绝对增量
- 等于 rate() * 时间范围秒数
- 适合计算总量变化
4. 实际应用示例
# QPS（每秒查询数）
sum(rate(http_requests_total[5m])) by (job)
# 错误率
sum(rate(http_requests_total{status=~"5.."}[5m])) /
sum(rate(http_requests_total[5m]))
# CPU使用率
100 - (avg(irate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)
# 网络流量
rate(node_network_receive_bytes_total[5m]) * 8 # 接收比特率
rate(node_network_transmit_bytes_total[5m]) * 8 # 发送比特率
"""
def explain_aggregation_functions(self) -> str:
"""解释聚合函数"""
return """
聚合函数：
1. sum() - 求和
sum(http_requests_total) # 所有请求总数
sum(http_requests_total) by (job) # 按job分组求和
sum(http_requests_total) without (instance) # 除instance外聚合
2. avg() - 平均值
avg(cpu_usage_percent) # 平均CPU使用率
avg(cpu_usage_percent) by (instance) # 按实例平均
3. min() / max() - 最小值/最大值
min(memory_usage_percent) # 最小内存使用率
max(disk_usage_percent) by (device) # 按设备最大磁盘使用率
4. count() - 计数
count(up) # 监控目标总数
count(up == 1) # 在线目标数
count(cpu_usage_percent > 80) # 高CPU使用率实例数
5. stddev() / stdvar() - 标准差/方差
stddev(response_time_seconds) # 响应时间标准差
stdvar(cpu_usage_percent) # CPU使用率方差
6. topk() / bottomk() - 前k个/后k个
topk(5, cpu_usage_percent) # CPU使用率最高的5个实例
bottomk(3, memory_available_bytes) # 可用内存最少的3个实例
7. quantile() - 分位数
quantile(0.95, response_time_seconds) # 95%分位数响应时间
quantile(0.5, cpu_usage_percent) # 中位数CPU使用率
8. 实际应用示例
# 集群总QPS
sum(rate(http_requests_total[5m]))
# 平均响应时间
avg(histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])))
# 服务可用性
avg(up) * 100
# 异常实例数量
count(up == 0)
# 资源使用率分布
histogram_quantile(0.95,
sum(rate(cpu_usage_bucket[5m])) by (le)
)
"""
def explain_time_functions(self) -> str:
"""解释时间函数"""
return """
时间函数：
1. time() - 当前时间戳
time() # 返回当前Unix时间戳
# 计算运行时间
time() - process_start_time_seconds
2. timestamp() - 样本时间戳
timestamp(up) # 返回up指标的时间戳
3. day_of_month() / day_of_week() - 日期函数
day_of_month() # 当前月份中的天数 (1-31)
day_of_week() # 当前星期几 (0=Sunday, 6=Saturday)
# 工作日过滤
cpu_usage_percent and on() (day_of_week() > 0 and day_of_week() < 6)
4. hour() / minute() - 时间函数
hour() # 当前小时 (0-23)
minute() # 当前分钟 (0-59)
# 工作时间过滤
cpu_usage_percent and on() (hour() >= 9 and hour() <= 17)
5. month() / year() - 日期函数
month() # 当前月份 (1-12)
year() # 当前年份
6. 实际应用示例
# 工作时间告警
(
cpu_usage_percent > 80
) and on() (
hour() >= 9 and hour() <= 17 and
day_of_week() > 0 and day_of_week() < 6
)
# 服务运行时间
(time() - process_start_time_seconds) / 3600 # 小时
# 按时间段分析
avg_over_time(
cpu_usage_percent[1h] @ (time() - 3600) # 1小时前的平均值
)
"""
def explain_math_functions(self) -> str:
"""解释数学函数"""
return """
数学函数：
1. abs() - 绝对值
abs(temperature_celsius - 20) # 与20度的温差
2. ceil() / floor() / round() - 取整函数
ceil(cpu_usage_percent) # 向上取整
floor(memory_usage_gb) # 向下取整
round(response_time_ms) # 四舍五入
3. sqrt() / exp() / ln() / log2() / log10() - 数学运算
sqrt(variance_value) # 平方根
ln(growth_rate) # 自然对数
log10(request_count) # 常用对数
4. clamp_max() / clamp_min() - 限制函数
clamp_max(cpu_usage_percent, 100) # 限制最大值为100
clamp_min(memory_free_bytes, 0) # 限制最小值为0
5. 三角函数
sin() / cos() / tan() / asin() / acos() / atan()
# 生成周期性测试数据
sin(time() / 3600 * 2 * 3.14159) # 1小时周期的正弦波
6. 实际应用示例
# 计算标准化分数
(
cpu_usage_percent - avg(cpu_usage_percent)
) / stddev(cpu_usage_percent)
# 计算变化率的绝对值
abs(rate(metric_value[5m]))
# 限制告警阈值
clamp_max(
predict_linear(disk_usage_percent[1h], 3600),
100
)
"""
class PromQLAdvanced:
"""PromQL高级用法"""
def explain_subqueries(self) -> str:
"""解释子查询"""
return """
子查询：
语法：<instant_query>[<range>:<resolution>]
1. 基本子查询
# 计算过去1小时内每5分钟的平均CPU使用率
avg_over_time(cpu_usage_percent[5m])[1h:5m]
# 计算过去1天内每小时的最大请求率
max_over_time(rate(http_requests_total[5m])[1h:])[1d:1h]
2. 复杂子查询
# 计算CPU使用率的变化趋势
deriv(
avg_over_time(cpu_usage_percent[5m])[1h:5m]
)
# 计算请求率的标准差
stddev_over_time(
rate(http_requests_total[5m])[1h:1m]
)
3. 实际应用场景
# 检测异常峰值
(
max_over_time(cpu_usage_percent[5m])[1h:5m] -
avg_over_time(cpu_usage_percent[5m])[1h:5m]
) > 20
# 计算SLA
avg_over_time(
(rate(http_requests_total{status="200"}[5m]) /
rate(http_requests_total[5m]))[1d:5m]
) * 100
注意事项：
- 子查询会增加计算复杂度
- 合理设置分辨率避免性能问题
- 考虑数据保留期限
"""
def explain_histogram_quantile(self) -> str:
"""解释直方图分位数"""
return """
histogram_quantile() 函数：
语法：histogram_quantile(φ, b)
- φ: 分位数 (0 ≤ φ ≤ 1)
- b: 直方图桶的时间序列
1. 基本用法
# 95%分位数响应时间
histogram_quantile(0.95,
rate(http_request_duration_seconds_bucket[5m])
)
# 99%分位数请求大小
histogram_quantile(0.99,
rate(http_request_size_bytes_bucket[5m])
)
2. 按标签分组
# 按服务计算95%分位数
histogram_quantile(0.95,
sum(rate(http_request_duration_seconds_bucket[5m])) by (le, service)
)
# 按方法和状态码计算
histogram_quantile(0.9,
sum(rate(http_request_duration_seconds_bucket[5m])) by (le, method, status)
)
3. 多分位数比较
# P50, P95, P99对比
histogram_quantile(0.5, sum(rate(http_request_duration_seconds_bucket[5m])) by (le))
histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[5m])) by (le))
histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[5m])) by (le))
4. SLA计算
# 计算99%的请求在1秒内完成的比例
(
histogram_quantile(0.99,
sum(rate(http_request_duration_seconds_bucket[5m])) by (le)
) < 1
) * 100
5. 告警规则
# P95响应时间告警
histogram_quantile(0.95,
sum(rate(http_request_duration_seconds_bucket[5m])) by (le, job)
) > 0.5
注意事项：
- 需要包含le标签
- 桶边界要合理设计
- 聚合时保留le标签
- 分位数是近似值
"""
def create_complex_examples(self) -> Dict[str, str]:
"""创建复杂查询示例"""
return {
"服务可用性SLA": """
# 计算过去24小时的服务可用性
(
sum(rate(http_requests_total{status=~"2..|3.."}[5m]))
/
sum(rate(http_requests_total[5m]))
) * 100
""",
"错误率趋势": """
# 计算错误率的1小时移动平均
avg_over_time(
(
sum(rate(http_requests_total{status=~"5.."}[5m]))
/
sum(rate(http_requests_total[5m]))
)[1h:5m]
) * 100
""",
"资源使用预测": """
# 预测1小时后的磁盘使用率
predict_linear(
(
node_filesystem_size_bytes - node_filesystem_free_bytes
) / node_filesystem_size_bytes * 100[1h:],
3600
)
""",
"异常检测": """
# 检测CPU使用率异常（超过历史平均值2个标准差）
abs(
cpu_usage_percent -
avg_over_time(cpu_usage_percent[1h])
) > 2 * stddev_over_time(cpu_usage_percent[1h])
""",
"多维度聚合": """
# 按数据中心和服务类型聚合请求率
sum(
rate(http_requests_total[5m])
) by (datacenter, service_type)
""",
"容量规划": """
# 计算基于当前增长率的容量耗尽时间
(
node_filesystem_free_bytes /
(
-1 * deriv(
node_filesystem_free_bytes[1h]
)
)
) / 3600 # 转换为小时
""",
"性能基线": """
# 建立性能基线（工作时间的P95响应时间）
histogram_quantile(0.95,
sum(
rate(http_request_duration_seconds_bucket[5m])
and on() (
hour() >= 9 and hour() <= 17 and
day_of_week() > 0 and day_of_week() < 6
)
) by (le)
)
""",
"故障影响分析": """
# 计算故障期间的请求损失
sum(
(
avg_over_time(rate(http_requests_total[5m])[1h:5m] offset 1d) -
rate(http_requests_total[5m])
) > 0
) * 300 # 5分钟间隔转换为总请求数
"""
}
class PromQLOptimization:
"""PromQL查询优化"""
def explain_performance_tips(self) -> str:
"""解释性能优化技巧"""
return """
PromQL性能优化技巧：
1. 选择器优化
# 好的做法：使用具体的标签过滤
http_requests_total{job="api", instance="web-1"}
# 避免：过于宽泛的选择器
http_requests_total
2. 时间范围优化
# 好的做法：使用合适的时间范围
rate(http_requests_total[5m])
# 避免：过长的时间范围
rate(http_requests_total[1d])
3. 聚合优化
# 好的做法：先过滤再聚合
sum(rate(http_requests_total{status="200"}[5m])) by (job)
# 避免：先聚合再过滤
sum(rate(http_requests_total[5m])) by (job, status) and on() status == "200"
4. 函数使用优化
# 好的做法：使用irate()检测瞬时变化
irate(cpu_seconds_total[5m])
# 好的做法：使用rate()计算平均速率
rate(http_requests_total[5m])
5. 标签基数控制
# 避免：高基数标签
http_requests_total{user_id="12345"} # 用户ID基数太高
# 好的做法：使用低基数标签
http_requests_total{user_type="premium"}
6. 查询复杂度控制
# 避免：过于复杂的嵌套查询
# 好的做法：分解为多个简单查询
7. 缓存利用
# 使用recording rules预计算常用查询
# 避免重复计算相同的表达式
"""
def explain_recording_rules(self) -> str:
"""解释记录规则"""
return """
记录规则 (Recording Rules)：
目的：
- 预计算复杂查询
- 提高查询性能
- 减少实时计算负载
1. 基本语法
groups:
- name: example_rules
interval: 30s
rules:
- record: job:http_requests:rate5m
expr: sum(rate(http_requests_total[5m])) by (job)
2. 命名约定
level:metric:operations
示例：
- instance:cpu_utilization:rate5m
- job:http_requests:rate5m
- cluster:memory_usage:ratio
3. 实际示例
groups:
- name: node_rules
interval: 30s
rules:
# CPU使用率
- record: instance:cpu_utilization:rate5m
expr: >
100 - (
avg(irate(node_cpu_seconds_total{mode="idle"}[5m])) by (instance) * 100
)
# 内存使用率
- record: instance:memory_utilization:ratio
expr: >
(
node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes
) / node_memory_MemTotal_bytes
# 磁盘使用率
- record: instance:disk_utilization:ratio
expr: >
(
node_filesystem_size_bytes - node_filesystem_free_bytes
) / node_filesystem_size_bytes
4. 分层记录规则
# 第一层：基础指标
- record: instance:cpu_utilization:rate5m
expr: 100 - (avg(irate(node_cpu_seconds_total{mode="idle"}[5m])) by (instance) * 100)
# 第二层：聚合指标
- record: job:cpu_utilization:mean5m
expr: avg(instance:cpu_utilization:rate5m) by (job)
# 第三层：集群级指标
- record: cluster:cpu_utilization:mean5m
expr: avg(job:cpu_utilization:mean5m)
5. 最佳实践
- 使用一致的命名约定
- 合理设置计算间隔
- 避免过度预计算
- 定期清理无用规则
- 监控规则计算性能
"""
# 使用示例
basics = PromQLBasics()
operators = PromQLOperators()
functions = PromQLFunctions()
advanced = PromQLAdvanced()
optimization = PromQLOptimization()
# 基础概念
print(basics.explain_data_types())
print("\n" + "="*50)
print(basics.explain_selectors())
# 基础示例
examples = basics.create_basic_examples()
for category, example_list in examples.items():
print(f"\n{category}:")
for example in example_list:
print(f" {example}")
# 操作符
print("\n" + "="*50)
print(operators.explain_arithmetic_operators())
print("\n" + "="*50)
print(operators.explain_comparison_operators())
print("\n" + "="*50)
print(operators.explain_logical_operators())
# 函数
print("\n" + "="*50)
print(functions.explain_rate_functions())
print("\n" + "="*50)
print(functions.explain_aggregation_functions())
print("\n" + "="*50)
print(functions.explain_time_functions())
print("\n" + "="*50)
print(functions.explain_math_functions())
# 高级用法
print("\n" + "="*50)
print(advanced.explain_subqueries())
print("\n" + "="*50)
print(advanced.explain_histogram_quantile())
# 复杂示例
complex_examples = advanced.create_complex_examples()
print("\n复杂查询示例:")
for name, query in complex_examples.items():
print(f"\n{name}:")
print(query)
# 性能优化
print("\n" + "="*50)
print(optimization.explain_performance_tips())
print("\n" + "="*50)
print(optimization.explain_recording_rules())
实战练习
1. 基础查询练习
class PromQLExercises:
"""PromQL练习题"""
def __init__(self):
self.exercises = {}
def basic_exercises(self) -> Dict[str, Dict[str, str]]:
"""基础练习题"""
return {
"练习1": {
"题目": "查询所有Prometheus实例的up状态",
"答案": "up{job='prometheus'}",
"解释": "使用标签过滤器选择特定job的up指标"
},
"练习2": {
"题目": "查询过去5分钟内HTTP请求的平均每秒速率",
"答案": "rate(http_requests_total[5m])",
"解释": "使用rate()函数计算Counter指标的每秒增长率"
},
"练习3": {
"题目": "查询CPU使用率大于80%的实例",
"答案": "cpu_usage_percent > 80",
"解释": "使用比较操作符过滤满足条件的时间序列"
},
"练习4": {
"题目": "计算所有实例的平均内存使用率",
"答案": "avg(memory_usage_percent)",
"解释": "使用avg()聚合函数计算所有时间序列的平均值"
},
"练习5": {
"题目": "查询method为GET或POST的HTTP请求",
"答案": "http_requests_total{method=~'GET|POST'}",
"解释": "使用正则匹配操作符=~匹配多个值"
}
}
def intermediate_exercises(self) -> Dict[str, Dict[str, str]]:
"""中级练习题"""
return {
"练习1": {
"题目": "计算HTTP 5xx错误率",
"答案": "sum(rate(http_requests_total{status=~'5..'}[5m])) / sum(rate(http_requests_total[5m]))",
"解释": "错误请求率除以总请求率"
},
"练习2": {
"题目": "按job分组计算CPU使用率最高的3个实例",
"答案": "topk(3, cpu_usage_percent) by (job)",
"解释": "使用topk()函数获取每个job中CPU使用率最高的实例"
},
"练习3": {
"题目": "计算磁盘使用率并转换为百分比",
"答案": "(node_filesystem_size_bytes - node_filesystem_free_bytes) / node_filesystem_size_bytes * 100",
"解释": "(总空间-可用空间)/总空间*100"
},
"练习4": {
"题目": "查询1小时前的CPU使用率",
"答案": "cpu_usage_percent offset 1h",
"解释": "使用offset修饰符查询历史数据"
},
"练习5": {
"题目": "计算HTTP请求的95%分位数响应时间",
"答案": "histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))",
"解释": "使用histogram_quantile()函数计算直方图分位数"
}
}
def advanced_exercises(self) -> Dict[str, Dict[str, str]]:
"""高级练习题"""
return {
"练习1": {
"题目": "检测CPU使用率异常峰值（超过1小时平均值的2倍标准差）",
"答案": "abs(cpu_usage_percent - avg_over_time(cpu_usage_percent[1h])) > 2 * stddev_over_time(cpu_usage_percent[1h])",
"解释": "使用统计函数检测异常值"
},
"练习2": {
"题目": "预测基于当前趋势1小时后的磁盘使用率",
"答案": "predict_linear(disk_usage_percent[1h], 3600)",
"解释": "使用predict_linear()函数进行线性预测"
},
"练习3": {
"题目": "计算工作时间内的平均响应时间",
"答案": "avg(response_time_seconds and on() (hour() >= 9 and hour() <= 17 and day_of_week() > 0 and day_of_week() < 6))",
"解释": "结合时间函数过滤工作时间数据"
},
"练习4": {
"题目": "计算服务的可用性SLA（99.9%）",
"答案": "(sum(rate(http_requests_total{status!~'5..'}[5m])) / sum(rate(http_requests_total[5m]))) * 100 > 99.9",
"解释": "计算非5xx请求占总请求的比例"
},
"练习5": {
"题目": "使用子查询计算过去24小时内每小时的最大QPS",
"答案": "max_over_time(sum(rate(http_requests_total[5m]))[1h:])[24h:1h]",
"解释": "使用子查询进行时间窗口聚合"
}
}
def create_practice_scenarios(self) -> Dict[str, str]:
"""创建实践场景"""
return {
"Web服务监控": """
# 场景：监控Web服务的关键指标
# 1. QPS（每秒请求数）
sum(rate(http_requests_total[5m])) by (service)
# 2. 错误率
sum(rate(http_requests_total{status=~"5.."}[5m])) by (service) /
sum(rate(http_requests_total[5m])) by (service) * 100
# 3. 平均响应时间
sum(rate(http_request_duration_seconds_sum[5m])) by (service) /
sum(rate(http_request_duration_seconds_count[5m])) by (service)
# 4. P95响应时间
histogram_quantile(0.95,
sum(rate(http_request_duration_seconds_bucket[5m])) by (le, service)
)
# 5. 服务可用性
avg(up{job=~"web.*"}) by (service) * 100
""",
"系统资源监控": """
# 场景：监控系统资源使用情况
# 1. CPU使用率
100 - (avg(irate(node_cpu_seconds_total{mode="idle"}[5m])) by (instance) * 100)
# 2. 内存使用率
(node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes) /
node_memory_MemTotal_bytes * 100
# 3. 磁盘使用率
(node_filesystem_size_bytes - node_filesystem_free_bytes) /
node_filesystem_size_bytes * 100
# 4. 网络流量
rate(node_network_receive_bytes_total[5m]) * 8 # 接收比特率
rate(node_network_transmit_bytes_total[5m]) * 8 # 发送比特率
# 5. 负载平均值
node_load1 # 1分钟负载
node_load5 # 5分钟负载
node_load15 # 15分钟负载
""",
"数据库监控": """
# 场景：监控MySQL数据库性能
# 1. 连接数
mysql_global_status_threads_connected
# 2. QPS
rate(mysql_global_status_queries[5m])
# 3. 慢查询率
rate(mysql_global_status_slow_queries[5m]) /
rate(mysql_global_status_queries[5m]) * 100
# 4. InnoDB缓冲池命中率
(mysql_global_status_innodb_buffer_pool_read_requests -
mysql_global_status_innodb_buffer_pool_reads) /
mysql_global_status_innodb_buffer_pool_read_requests * 100
# 5. 复制延迟
mysql_slave_lag_seconds
""",
"容器监控": """
# 场景：监控Kubernetes容器
# 1. Pod CPU使用率
sum(rate(container_cpu_usage_seconds_total{container!="POD"}[5m])) by (pod) * 100
# 2. Pod内存使用率
sum(container_memory_working_set_bytes{container!="POD"}) by (pod) /
sum(container_spec_memory_limit_bytes{container!="POD"}) by (pod) * 100
# 3. Pod重启次数
increase(kube_pod_container_status_restarts_total[1h])
# 4. 节点资源使用率
sum(rate(container_cpu_usage_seconds_total[5m])) by (node) /
sum(kube_node_status_allocatable{resource="cpu"}) by (node) * 100
# 5. 集群Pod数量
sum(kube_pod_info) by (namespace)
"""
}
# 使用示例
exercises = PromQLExercises()
# 基础练习
basic = exercises.basic_exercises()
print("基础练习题:")
for name, exercise in basic.items():
print(f"\n{name}:")
print(f"题目: {exercise['题目']}")
print(f"答案: {exercise['答案']}")
print(f"解释: {exercise['解释']}")
# 中级练习
intermediate = exercises.intermediate_exercises()
print("\n" + "="*50)
print("中级练习题:")
for name, exercise in intermediate.items():
print(f"\n{name}:")
print(f"题目: {exercise['题目']}")
print(f"答案: {exercise['答案']}")
print(f"解释: {exercise['解释']}")
# 高级练习
advanced = exercises.advanced_exercises()
print("\n" + "="*50)
print("高级练习题:")
for name, exercise in advanced.items():
print(f"\n{name}:")
print(f"题目: {exercise['题目']}")
print(f"答案: {exercise['答案']}")
print(f"解释: {exercise['解释']}")
# 实践场景
scenarios = exercises.create_practice_scenarios()
print("\n" + "="*50)
print("实践场景:")
for scenario_name, queries in scenarios.items():
print(f"\n{scenario_name}:")
print(queries)
总结
关键要点
PromQL语法掌握
- 四种数据类型理解
- 选择器和匹配器使用
- 时间范围和偏移修饰符
- 操作符优先级和结合性
函数分类应用
- 速率函数：rate(), irate(), increase()
- 聚合函数：sum(), avg(), max(), min()
- 时间函数：time(), hour(), day_of_week()
- 数学函数：abs(), sqrt(), clamp_max()
高级特性运用
- 子查询进行复杂时间窗口分析
- histogram_quantile()计算分位数
- 向量匹配和标签操作
- 预测和异常检测函数
性能优化策略
- 合理使用标签过滤
- 控制查询时间范围
- 利用recording rules预计算
- 避免高基数标签
最佳实践
查询设计
- 先过滤后聚合
- 使用具体的标签选择器
- 合理设置时间范围
- 避免过于复杂的嵌套
性能考虑
- 监控查询执行时间
- 使用recording rules优化
- 控制返回的时间序列数量
- 定期清理无用指标
可维护性
- 使用有意义的指标名称
- 添加适当的注释
- 建立查询模板库
- 文档化复杂查询逻辑
下一步学习建议
深入实践
- 在真实环境中练习查询
- 分析不同场景的监控需求
- 建立个人查询库
扩展学习
- 学习Grafana中的PromQL应用
- 掌握告警规则编写
- 了解Federation和远程存储
性能调优
- 学习Prometheus内部机制
- 掌握TSDB优化技巧
- 监控Prometheus自身性能
生态集成
- 学习各种Exporter的指标
- 掌握服务发现配置
- 了解云原生监控最佳实践
通过本章学习，你应该能够： - 熟练编写各种PromQL查询 - 理解不同函数的适用场景 - 优化查询性能和可维护性 - 解决实际监控场景中的问题
继续学习后续章节，你将能够构建完整的监控解决方案！
