PromQL查询语言和日常使用的数据库SQL查询语言（SELECT * FROM ...）是不同的，PromQL 是一种嵌套的函数式语言，就是我们要把需要查找的数据描述成一组嵌套的表达式，每个表达式都会评估为一个中间值，每个中间值都会被用作它上层表达式中的参数，而查询的最外层表达式表示你可以在表格、图形中看到的最终返回值。

比如下面的查询语句：


histogram_quantile( # 查询的根，最终结果表示一个近似分位数。
0.9, # histogram_quantile() 的第一个参数，分位数的目标值
# histogram_quantile() 的第二个参数，聚合的直方图
sum by(le, method, path) (
# sum() 的参数，直方图过去5分钟每秒增量。
rate(
# rate() 的参数，过去5分钟的原始直方图序列
demo_api_request_duration_seconds_bucket{job="demo"}[5m]
)
)
)

然后还需要认识一下告警的 PromQL 中，经常出现的一些函数：

rate

用于计算变化率的最常见函数是 rate()，rate()函数用于计算在指定时间范围内计数器平均每秒的增加量。因为是计算一个时间范围内的平均值，所以我们需要在序列选择器之后添加一个范围选择器。

irate

由于使用 rate 或者 increase 函数去计算样本的平均增长速率，容易陷入长尾问题当中，其无法反应在时间窗口内样本数据的突发变化。

例如，对于主机而言在 2 分钟的时间窗口内，可能在某一个由于访问量或者其它问题导致 CPU 占用 100%的情况，但是通过计算在时间窗口内的平均增长率却无法反应出该问题。

为了解决该问题，PromQL 提供了另外一个灵敏度更高的函数irate(v range-vector)。irate 同样用于计算区间向量的计算率，但是其反应出的是瞬时增长率。

histogram_quantile

获取数据的分位数。histogram_quantile(φ scalar, b instant-vector) 函数用于计算历史数据指标一段时间内的分位数。该函数将目标分位数 (0 ≤ φ ≤ 1) 和直方图指标作为输入，就是大家平时讲的 pxx，p50 就是中位数，参数 b 一定是包含 le 这个标签的瞬时向量，不包含就无从计算分位数了，但是计算的分位数是一个预估值，并不完全准确，因为这个函数是假定每个区间内的样本分布是线性分布来计算结果值的，预估的准确度取决于 bucket 区间划分的粒度，粒度越大，准确度越低。

该部分引用：Prometheus基础相关--PromQL 基础(2)

（Prometheus基础相关--PromQL 基础(2) - 知乎）想学习的同学可以去看看原文

三、修改PromQL

要让巡检使用 PromQL ，就必须要修改告警中的 PromQL。这里需要介绍一个函数：max_over_time(range-vector)，它是获取区间向量内每个指标的最大值。其实还有其他这类时间聚合函数，比如avg_over_time、min_over_time、sum_over_time等等，但是我们只需要获取到最大值，来提醒 dba 就行了。

Prometheus 是支持子查询的，它允许我们首先以指定的步长在一段时间内执行内部查询，然后根据子查询的结果计算外部查询。子查询的表示方式类似于区间向量的持续时间，但需要冒号后添加了一个额外的步长参数：[<duration>:<resolution>]。

举个例子：


# 原版
sum(rate(tikv_thread_cpu_seconds_total{name=~"(raftstore|rs)_.*"}[1m])) by(instance)
 
# 修改
max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"(raftstore|rs)_.*"}[1m])) by(instance)[24h:1m])

这是获取 TiKV raftstore 线程池 CPU 使用率的告警项。原版是直接将1分钟内所有线程的变化率相加，而笔者的修改版是将1分钟内所有线程的使用率取平均值，并且从此刻向后倒24小时内，每一分钟执行一次获取平均线程使用率的查询，再取最大值。

也就是说，从24小时前，到现在，每分钟执行一次（步长为1分钟）：avg(rate(tikv_thread_cpu_seconds_total{name=~"(raftstore|rs)_.*"}[1m])) by (instance)，并获取其中最大的一次值。这样就满足了我们需要使用一段时间的数据来判断集群是否有风险的依据了。

然后我们可以选取合适的 PromQL 来加上时间聚合函数和查询时间及步长信息：


# TiKV 1
'TiDB.tikv.TiKV_server_is_down': {
'pql': 'probe_success{group="tikv",instance=~".*"} == 0',
'pql_max': '',
'note': 'TiKV 服务不可用'
},
'TiDB.tikv.TiKV_node_restart': {
'pql': 'changes(process_start_time_seconds{job="tikv",instance=~".*"}[24h])> 0',
'pql_max': 'max(changes(process_start_time_seconds{job="tikv",instance=~".*"}[24h]))',
'note': 'TiKV 服务5分钟内出现重启'
},
'TiDB.tikv.TiKV_GC_can_not_work': {
'pql_max': '',
'pql': 'sum(increase(tikv_gcworker_gc_tasks_vec{task="gc", instance=~".*"}[2d])) by (instance) < 1and (sum(increase('
'tikv_gc_compaction_filter_perform{instance=~".*"}[2d])) by (instance) < 1and sum(increase('
'tikv_engine_event_total{cf="write",db="kv",type="compaction",instance=~".*"}[2d])) by (instance) >= 1)',
'note': 'TiKV 服务GC无法工作'
},
# TiKV 2
'TiDB.tikv.TiKV_raftstore_thread_cpu_seconds_total': {
'pql_max': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"(raftstore|rs)_.*"}[1m])) by (instance)[24h:1m])',
'pql': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"(raftstore|rs)_.*"}[1m])) by (instance)[24h:1m]) > 0.8',
'note': 'TiKV raftstore 线程池 CPU 使用率过高'
},
'TiDB.tikv.TiKV_approximate_region_size': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_region_size_bucket{instance=~".*"}[1m])) '
'by (le,instance))[24h:1m])',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_region_size_bucket{instance=~".*"}[1m])) '
'by (le,instance))[24h:1m]) > 1073741824',
'note': 'TiKV split checker 扫描到的最大的 Region approximate size 大于 1GB'
},
'TiDB.tikv.TiKV_async_request_write_duration_seconds': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_storage_engine_async_request_duration_seconds_bucket'
'{type="write", instance=~".*"}[1m])) by (le, instance, type))[24h:1m])',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_storage_engine_async_request_duration_seconds_bucket'
'{type="write", instance=~".*"}[1m])) by (le, instance, type))[24h:1m]) > 1',
'note': 'TiKV 中Raft写入响应时间过长'
},
'TiDB.tikv.TiKV_scheduler_command_duration_seconds': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_scheduler_command_duration_seconds_bucket[20m])) by (le, instance, type) / 1000)[24h:20m]) ',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_scheduler_command_duration_seconds_bucket[20m])) by (le, instance, type) / 1000)[24h:20m]) > 20',
'note': 'TiKV 调度器请求响应时间过长'
},
'TiDB.tikv.TiKV_scheduler_latch_wait_duration_seconds': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_scheduler_latch_wait_duration_seconds_bucket[20m])) by (le, instance, type))[24h:20m]) ',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_scheduler_latch_wait_duration_seconds_bucket[20m])) by (le, instance, type))[24h:20m]) > 20',
'note': 'TiKV 调度器锁等待响应时间过长'
},
'TiDB.tikv.TiKV_write_stall': {
'pql_max': 'max_over_time(delta(tikv_engine_write_stall{instance=~".*"}[10m])[24h:10m])',
'pql': 'max_over_time(delta('
'tikv_engine_write_stall{instance=~".*"}[10m])[24h:10m]) > 10',
'note': 'TiKV 中存在写入积压'
},
 
# TiKV 3
'TiDB.tikv.TiKV_server_report_failure_msg_total': {
'pql_max': 'max_over_time(sum(rate(tikv_server_report_failure_msg_total{type="unreachable"}[10m])) BY (instance)[24h:10m])',
'pql': 'max_over_time(sum(rate(tikv_server_report_failure_msg_total{type="unreachable"}[10m])) BY (instance)[24h:10m]) > 10',
'note': 'TiKV 节点报告失败次数过多'
},
'TiDB.tikv.TiKV_channel_full_total': {
'pql_max': 'max_over_time(sum(rate(tikv_channel_full_total{instance=~".*"}[10m])) BY (type, instance)[24h:10m])',
'pql': 'max_over_time(sum(rate(tikv_channel_full_total{instance=~".*"}[10m])) BY (type, instance)[24h:10m]) > 0',
'note': 'TIKV 通道已占满 tikv 过忙'
},
'TiDB.tikv.TiKV_raft_log_lag': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_log_lag_bucket{instance=~".*"}[1m])) by (le,instance))[24h:10m])',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_log_lag_bucket{instance=~".*"}[1m])) by (le, '
'instance))[24h:10m]) > 5000',
'note': 'TiKV 中 raft 日志同步相差过大'
},
'TiDB.tikv.TiKV_thread_unified_readpool_cpu_seconds': {
'pql_max': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"unified_read_po*", instance=~".*"}[1m])) by (instance)[24h:1m])',
'pql': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"unified_read_po*", instance=~".*"}[1m])) '
'by (instance)[24h:1m]) > 0.7',
'note': 'unifiled read 线程池使用率大于70%'
},
'TiDB.tikv.TiKV_low_space': {
'pql_max': 'sum(tikv_store_size_bytes{type="available"}) by (instance) / sum(tikv_store_size_bytes{type="capacity"}) by (instance)',
'pql': 'sum(tikv_store_size_bytes{type="available"}) by (instance) / sum(tikv_store_size_bytes{type="capacity"}) by (instance) < 0.3',
'note': 'TiKV 当前存储可用空间小于阈值'
},

由于有的告警项是获取了5分钟或者10分钟的数据，在写步长的时候也要同步修改为5分钟或者10分钟，保持一致可以保证，检查能覆盖选定的全部时间段，并且不会重复计算造成资源浪费。

顺带一提，如果不加max_over_time 可以获取到带有时间戳的全部数据，而不是只获取到最大的一个数据。这个带时间戳的全部数据可以方便画图，像grafana那样展示数据趋势。

四、巡检脚本

了解了以上所有知识，我们就可以开始编写巡检脚本了。

这是笔者和同事共同编写的一部分巡检脚本，最重要的是 tasks 中的 PromQL ，在脚本执行之前要写好 PromQL，其他部分可以随意更改。如果一次性巡检天数太多，比如一次巡检一个月的时间，Prometheus 可能会因检查数据太多而报错的，所以使用的时候要注意报错信息，避免漏掉一些巡检项。


# -*- coding: utf-8 -*-
importsubprocess
importre
importdatetime
importrequests
importsys
importpandas aspd
 
days = None
 
 
defget_cluster_name():
try:
command = "tiup cluster list"
result = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, error = result.communicate()
cluster_name_match = re.search(r'([a-zA-Z0-9_-]+)\s+tidb\s+v', output.decode('utf-8'))
ifcluster_name_match:
returncluster_name_match.group(1)
else:
returnNone
exceptException ase:
print("An error occurred:", e)
returnNone
 
 
defdisplay_cluster_info(cluster_name):
ifnotcluster_name:
print("Cluster name not found.")
return
 
try:
command = "tiup cluster display {0}".format(cluster_name)
result = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, error = result.communicate()
returnoutput.decode('utf-8')
exceptException ase:
print("An error occurred:", e)
 
 
defextract_id_role(output):
id_role_dict = {}
lines = output.strip().split("\n")
forline inlines:
print(line)
parts = line.split()
ifis_valid_ip_port(parts[0]):
node_id, role = parts[0], parts[1]
id_role_dict[node_id] = role
returnid_role_dict
 
 
defis_valid_ip_port(input_str):
pattern = re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d{1,5}$')
returnbool(pattern.match(input_str))
 
 
defget_prometheus_ip(data_dict):
prometheus_ip = None
forkey, value indata_dict.items():
ifvalue == 'prometheus':
prometheus_ip = key
break
returnprometheus_ip
 
 
defget_tasks():
globaldays
tasks = {
# TiKV 1
'TiDB.tikv.TiKV_server_is_down': {
'pql': 'probe_success{group="tikv",instance=~".*"} == 0',
'pql_max': '',
'note': 'TiKV 服务不可用'
},
'TiDB.tikv.TiKV_node_restart': {
'pql': 'changes(process_start_time_seconds{job="tikv",instance=~".*"}[24h])> 0',
'pql_max': 'max(changes(process_start_time_seconds{job="tikv",instance=~".*"}[24h]))',
'note': 'TiKV 服务5分钟内出现重启'
},
'TiDB.tikv.TiKV_GC_can_not_work': {
'pql_max': '',
'pql': 'sum(increase(tikv_gcworker_gc_tasks_vec{task="gc", instance=~".*"}[2d])) by (instance) < 1 and (sum(increase('
'tikv_gc_compaction_filter_perform{instance=~".*"}[2d])) by (instance) < 1 and sum(increase('
'tikv_engine_event_total{cf="write",db="kv",type="compaction",instance=~".*"}[2d])) by (instance) >= 1)',
'note': 'TiKV 服务GC无法工作'
},
# TiKV 2
'TiDB.tikv.TiKV_raftstore_thread_cpu_seconds_total': {
'pql_max': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"(raftstore|rs)_.*"}[1m])) by (instance)[24h:1m])',
'pql': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"(raftstore|rs)_.*"}[1m])) by (instance)[24h:1m]) > 0.8',
'note': 'TiKV raftstore 线程池 CPU 使用率过高'
},
'TiDB.tikv.TiKV_approximate_region_size': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_region_size_bucket{instance=~".*"}[1m])) '
'by (le,instance))[24h:1m])',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_region_size_bucket{instance=~".*"}[1m])) '
'by (le,instance))[24h:1m]) > 1073741824',
'note': 'TiKV split checker 扫描到的最大的 Region approximate size 大于 1 GB'
},
'TiDB.tikv.TiKV_async_request_write_duration_seconds': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_storage_engine_async_request_duration_seconds_bucket'
'{type="write", instance=~".*"}[1m])) by (le, instance, type))[24h:1m])',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_storage_engine_async_request_duration_seconds_bucket'
'{type="write", instance=~".*"}[1m])) by (le, instance, type))[24h:1m]) > 1',
'note': 'TiKV 中Raft写入响应时间过长'
},
'TiDB.tikv.TiKV_write_stall': {
'pql_max': 'max_over_time(delta(tikv_engine_write_stall{instance=~".*"}[10m])[24h:10m])',
'pql': 'max_over_time(delta('
'tikv_engine_write_stall{instance=~".*"}[10m])[24h:10m]) > 10',
'note': 'TiKV 中存在写入积压'
},
 
# TiKV 3
'TiDB.tikv.TiKV_server_report_failure_msg_total': {
'pql_max': 'max_over_time(sum(rate(tikv_server_report_failure_msg_total{type="unreachable"}[10m])) BY (instance)[24h:10m])',
'pql': 'max_over_time(sum(rate(tikv_server_report_failure_msg_total{type="unreachable"}[10m])) BY (instance)[24h:10m]) > 10',
'note': 'TiKV 节点报告失败次数过多'
},
'TiDB.tikv.TiKV_channel_full_total': {
'pql_max': 'max_over_time(sum(rate(tikv_channel_full_total{instance=~".*"}[10m])) BY (type, instance)[24h:10m])',
'pql': 'max_over_time(sum(rate(tikv_channel_full_total{instance=~".*"}[10m])) BY (type, instance)[24h:10m]) > 0',
'note': 'TIKV 通道已占满 tikv 过忙'
},
'TiDB.tikv.TiKV_raft_log_lag': {
'pql_max': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_log_lag_bucket{instance=~".*"}[1m])) by (le,instance))[24h:10m])',
'pql': 'max_over_time(histogram_quantile(0.99, sum(rate(tikv_raftstore_log_lag_bucket{instance=~".*"}[1m])) by (le, '
'instance))[24h:10m]) > 5000',
'note': 'TiKV 中 raft 日志同步相差过大'
},
'TiDB.tikv.TiKV_thread_unified_readpool_cpu_seconds': {
'pql_max': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"unified_read_po*", instance=~".*"}[1m])) by (instance)[24h:1m])',
'pql': 'max_over_time(avg(rate(tikv_thread_cpu_seconds_total{name=~"unified_read_po*", instance=~".*"}[1m])) '
'by (instance)[24h:1m]) > 0.7',
'note': 'unifiled read 线程池使用率大于70%'
},
'TiDB.tikv.TiKV_low_space': {
'pql_max': 'sum(tikv_store_size_bytes{type="available"}) by (instance) / sum(tikv_store_size_bytes{type="capacity"}) by (instance)',
'pql': 'sum(tikv_store_size_bytes{type="available"}) by (instance) / sum(tikv_store_size_bytes{type="capacity"}) by (instance) < 0.3',
'note': 'TiKV 当前存储可用空间小于阈值'
},
}
forkey, value intasks.items():
forinner_key, inner_value invalue.items():
ifisinstance(inner_value, str) and'pql'ininner_key:
value[inner_key] = inner_value.replace("24h:", f"{24* days}h:").replace("[24h]", f"[{24* days}h]")
returntasks
 
 
defrequest_prome(prometheus_address, query):
try:
response = requests.get('http://%s/api/v1/query'% prometheus_address, params={'query': query})
returnresponse
except:
returnNone
 
 
defhas_response(prometheus_address, query):
response = request_prome(prometheus_address, query)
ifnotresponse:
returnFalse
try:
ifresponse.json()["data"]['result']:
returnTrue
else:
returnFalse
except:
returnFalse
 
 
defcheck_prome_alive(prometheus_address):
# dummy query is used to judge if prometheus is alive
dummy_query = 'probe_success{}'
returnhas_response(prometheus_address, dummy_query)
 
 
deffind_alive_prome(prometheus_addresses):
ifcheck_prome_alive(prometheus_addresses):
returnprometheus_addresses
returnNone
 
 
# ip:port -> ip_port
defdecode_instance(instance):
returninstance.replace(':', '_')
 
 
defcheck_metric(alert_name, prometheus_address, pql, is_value, pql_max):
record = []
try:
is_warning = "异常"
response = request_prome(prometheus_address, pql)
alert_name = alert_name.split('.')
result = response.json()['data']['result']
 
# 判断是否出现异常
iflen(result) == 0:
is_warning = "正常"
ifpql_max == '':
result = [{'metric': {}, 'value': [0, '0']}]
else:
response = request_prome(prometheus_address, pql_max)
result = response.json()['data']['result']
 
fori inresult:
# 判断是否按节点显示
if'instance'ini['metric']:
instance = i['metric']['instance']
node = decode_instance(instance)
else:
node = '集群'
# 判断是否有type
if'type'ini['metric']:
type = i['metric']['type']
else:
type = '无类型'
value = i['value'][1]
 
ifvalue == 'NaN':
value = 0
else:
value = round(float(value), 3)
message = "%s,%s,%s,%s,%s,%s,%s,%s"% (
datetime.datetime.now(), node, alert_name[1], alert_name[2], type, is_warning, is_value, value)
print(message)
record.append(message)
exceptException ase:
print(alert_name[2] + "----An error occurred check_metric:", e)
return
returnrecord
 
 
defcsv_report(record):
data = pd.DataFrame([line.split(',') forline inrecord],
columns=['timestamp', 'ip_address', 'service', 'event_type', 'type', 'status', 'description',
'value'])
grouped = data.groupby("service")
writer = pd.ExcelWriter("inspection_report.xlsx", engine="xlsxwriter")
forname, group ingrouped:
group.to_excel(writer, sheet_name=name, index=False)
worksheet = writer.sheets[name]
fori, col inenumerate(group.columns):
column_len = max(group[col].astype(str).str.len().max(), len(col)) + 2
worksheet.set_column(i, i, column_len)
writer.save()
 
 
defrun_tasks(role_metrics, prometheus_address):
record = []
foralert inrole_metrics:
pql = role_metrics[alert]['pql']
is_value = role_metrics[alert]['note']
pql_max = role_metrics[alert]['pql_max']
message = check_metric(alert, prometheus_address, pql, is_value, pql_max)
fordata inmessage:
record.append(data)
csv_report(record)
 
 
defrun_script(prometheus_addresses):
active_prometheus_address = find_alive_prome(prometheus_addresses)
 
# check if all prometheus are down
ifnotactive_prometheus_address:
sys.exit()
tasks = get_tasks()
run_tasks(tasks, active_prometheus_address)
 
 
defget_user_input():
globaldays
try:
user_input = int(input("请输入需要巡检的天数: "))
days = user_input
exceptValueError:
print("输入无效，请输入一个有效的数字。")
 
 
if__name__ == "__main__":
# 输入巡检天数
get_user_input()
 
prometheus_ip = '10.3.65.136:9091'
# prometheus_ip = None
ifprometheus_ip isNone:
cluster_name = get_cluster_name()
cluster_info = display_cluster_info(cluster_name)
id_role_dict = extract_id_role(cluster_info)
print(id_role_dict)
prometheus_ip = get_prometheus_ip(id_role_dict)
print(prometheus_ip)
run_script(prometheus_ip)

五、总结

一个完善的巡检脚本的编写是一个长期的工作。因为时间有限，笔者只编写了基于 Prometheus 的一部分巡检项，有兴趣的同学可以继续编写更多巡检项。

目前巡检脚本都是基于 Prometheus 的数据来作判断，但是在真实的巡检当中，dba还会查看一些 Prometheus 没有的数据，比如表的健康度、一段时间内的慢SQL、热力图、日志信息等等，这些信息在后面一些时间，可能会慢慢入到巡检脚本中。

最后，笔者能力有限，如有错误的地方欢迎指出。

作者：蔡一凡| 后端开发工程师

公众号搜索神州数码云基地，后台回复数据库，加入数据库技术交流群。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小丑西瓜9/article/detail/287409