背景
笔者最近在驻场，发现这里的 tidb 集群是真的多，有将近150套集群。而且集群少则6个节点起步，多则有200多个节点。在这么庞大的集群体量下，巡检就变得非常的繁琐了。
那么有没有什么办法能够代替手动巡检，并且能够快速准确的获取到集群相关信息的方法呢？答案是，有但不完全有。其实可以利用 tidb 的 Prometheus 来获取集群相关的各项数据，比如告警就是一个很好的例子。可惜了，告警只是获取了当前数据进行告警判断，而巡检需要使用一段时间的数据来作为判断的依据。而且，告警是已经达到临界值了，巡检却是要排查集群的隐患，提前开始规划，避免出现异常。
那直接用 Prometheus 获取一段时间的数据，并且把告警值改低不就行了？
认识 PromQL
要使用 Prometheus ，那必须要先了解什么是 PromQL 。
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) 想学习的同学可以去看看原文
修改 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) < 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_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 -*-
import subprocess
import re
import datetime
import requests
import sys
import pandas as pd
days = None
def get_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'))
if cluster_name_match:
return cluster_name_match.group(1)
else:
return None
except Exception as e:
print("An error occurred:", e)
return None
def display_cluster_info(cluster_name):
if not cluster_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()
return output.decode('utf-8')
except Exception as e:
print("An error occurred:", e)
def extract_id_role(output):
id_role_dict = {}
lines = output.strip().split("\n")
for line in lines:
print(line)
parts = line.split()
if is_valid_ip_port(parts[0]):
node_id, role = parts[0], parts[1]
id_role_dict[node_id] = role
return id_role_dict
def is_valid_ip_port(input_str):
pattern = re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d{1,5}$')
return bool(pattern.match(input_str))
def get_prometheus_ip(data_dict):
prometheus_ip = None
for key, value in data_dict.items():
if value == 'prometheus':
prometheus_ip = key
break
return prometheus_ip
def get_tasks():
global days
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 当前存储可用空间小于阈值'
},
}
for key, value in tasks.items():
for inner_key, inner_value in value.items():
if isinstance(inner_value, str) and 'pql' in inner_key:
value[inner_key] = inner_value.replace("24h:", f"{24 * days}h:").replace("[24h]", f"[{24 * days}h]")
return tasks
def request_prome(prometheus_address, query):
try:
response = requests.get('http://%s/api/v1/query' % prometheus_address, params={'query': query})
return response
except:
return None
def has_response(prometheus_address, query):
response = request_prome(prometheus_address, query)
if not response:
return False
try:
if response.json()["data"]['result']:
return True
else:
return False
except:
return False
def check_prome_alive(prometheus_address):
# dummy query is used to judge if prometheus is alive
dummy_query = 'probe_success{}'
return has_response(prometheus_address, dummy_query)
def find_alive_prome(prometheus_addresses):
if check_prome_alive(prometheus_addresses):
return prometheus_addresses
return None
# ip:port -> ip_port
def decode_instance(instance):
return instance.replace(':', '_')
def check_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']
# 判断是否出现异常
if len(result) == 0:
is_warning = "正常"
if pql_max == '':
result = [{'metric': {}, 'value': [0, '0']}]
else:
response = request_prome(prometheus_address, pql_max)
result = response.json()['data']['result']
for i in result:
# 判断是否按节点显示
if 'instance' in i['metric']:
instance = i['metric']['instance']
node = decode_instance(instance)
else:
node = '集群'
# 判断是否有type
if 'type' in i['metric']:
type = i['metric']['type']
else:
type = '无类型'
value = i['value'][1]
if value == '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)
except Exception as e:
print(alert_name[2] + "----An error occurred check_metric:", e)
return
return record
def csv_report(record):
data = pd.DataFrame([line.split(',') for line in record],
columns=['timestamp', 'ip_address', 'service', 'event_type', 'type', 'status', 'description',
'value'])
grouped = data.groupby("service")
writer = pd.ExcelWriter("inspection_report.xlsx", engine="xlsxwriter")
for name, group in grouped:
group.to_excel(writer, sheet_name=name, index=False)
worksheet = writer.sheets[name]
for i, col in enumerate(group.columns):
column_len = max(group[col].astype(str).str.len().max(), len(col)) + 2
worksheet.set_column(i, i, column_len)
writer.save()
def run_tasks(role_metrics, prometheus_address):
record = []
for alert in role_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)
for data in message:
record.append(data)
csv_report(record)
def run_script(prometheus_addresses):
active_prometheus_address = find_alive_prome(prometheus_addresses)
# check if all prometheus are down
if not active_prometheus_address:
sys.exit()
tasks = get_tasks()
run_tasks(tasks, active_prometheus_address)
def get_user_input():
global days
try:
user_input = int(input("请输入需要巡检的天数: "))
days = user_input
except ValueError:
print("输入无效，请输入一个有效的数字。")
if __name__ == "__main__":
# 输入巡检天数
get_user_input()
prometheus_ip = '10.3.65.136:9091'
# prometheus_ip = None
if prometheus_ip is None:
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、热力图、日志信息等等，这些信息在后面一些时间，可能会慢慢入到巡检脚本中。
最后，笔者能力有限，如有错误的地方欢迎指出。
----------------------------------------------------
作者介绍：caiyfc，来自神州数码钛合金战队，是一支致力于为企业提供分布式数据库TiDB整体解决方案的专业技术团队。团队成员拥有丰富的数据库从业背景，全部拥有TiDB高级资格证书，并活跃于TiDB开源社区，是官方认证合作伙伴。目前已为10+客户提供了专业的TiDB交付服务，涵盖金融、证券、物流、电力、政府、零售等重点行业。