# coding=utf-8 # Usage: run_check.py base_path clsname # will read baseline.csv under base_path # will produce report_[datetime].csv under base_path # will read tidb_config under base_path/config # config format: # ENABLE:check_class:check_name:script_name:operator:threshold:check_args # printout format: # check_class check_name check_item colorized_check_status check_value operator+threshold # log/baseline: # 检查类别||检查项||检查对象||检查结果||检查值||检查阈值||出现时间 # check_class||check_name||check_item||check_status||check_value||operator+threshold||last_appearance # 导入模块 import os, sys, time # import requests # 定义变量 # 安装地址 base_path = sys.argv[1] # base_path = '/home/tidb/scripts/normalcheck' # 集群 cluster_name = sys.argv[2] # cluster_name = 'sa_drainer1' # 当前时间，格式为：年月日时分秒 current_time = os.popen('date "+%Y%m%d%H%M%S"').read().strip('\n') # 当前系统的名字 host_name = os.popen('hostname').read().strip('\n') # 配置文件地址 var_file = base_path + "/config/tidb_vars.config" config_file = base_path + "/config/tidb_check_item.config" # 检查脚本路径 script_dir = base_path + "/script" # 基准表的文件路径及名字 # baseline_file = base_path + "/config/baseline.csv" # 输出日志文件的路径及名字 output_log = base_path + "/output/tidb_normalcheck_report_" + host_name + "_" + cluster_name + "_" + current_time + ".csv" # 判断一个unicode是否是汉字 def is_chinese(uchar): if u'\u4e00' <= uchar <= u'\u9fa5': return True else: return False # 将输入文字填充到width长度，不会进行裁剪 # 对齐方式为左对齐 def format_text(text, width, colored=False): str_text = str(text) unicode_text = str_text.decode("utf-8") # unicode_text = str_text cn_count = 0 color_char = 0 if colored: color_char = 9 for u in unicode_text: if is_chinese(u): cn_count += 1 return str_text + (width + color_char - cn_count - len(unicode_text)) * " " # 屏幕输出字符宽度定义 a0 = 8 a1 = 30 a2 = 40 a3 = 8 a4 = 10 a5 = 15 a6 = 15 # 格式化各个指标并在屏幕输出 def format_line(str0, str1, str2, str3, str4, str5, str6, colored_result=False): fmt = "| {0} | {1} | {2} | {3} | {4} | {5} | {6} |" return fmt.format(format_text(str0, a0), format_text(str1, a1), format_text(str2, a2), format_text(str3, a3, colored_result), format_text(str4, a4), format_text(str5, a5), format_text(str6, a6) ) # 屏幕输出打印行间隔 def print_linebreak(): print(("+ {0:{0}<%d} + {0:{0}<%d} + {0:{0}<%d} + {0:{0}<%d} + {0:{0}<%d} + {0:{0}<%d} + {0:{0}<%d} + " % ( a0, a1, a2, a3, a4, a5, a6)).format('-')) # 屏幕输出标题格式定义 def print_title(): print_linebreak() print(format_line("检查类别", "检查项", "检查对象", "检查结果", "检查值", "检查阈值", "检查服务器")) print_linebreak() # 将一行的检查内容格式化并添加颜色，在屏幕输出 def print_text(check_class, check_name, check_item, check_status, check_value, operator, threshold, check_node): # 定义检查结果颜色 if check_status == "正常": check_status = "\033[32m%s\033[0m" % check_status else: check_status = "\033[31m%s\033[0m" % check_status print(format_line(check_class, check_name, check_item, check_status, check_value, operator + threshold, check_node, colored_result=True)) print_linebreak() # 向结果报告中写入一行 def write_log(check_class, check_name, check_item, check_status, check_value, operator, threshold, check_node, last_appearance): with open(output_log, 'a') as log: text = "%s||%s||%s||%s||%s||%s||%s||%s\n" % (check_class, check_name, check_item, check_status, check_value, operator + threshold, check_node, last_appearance) log.write(text) # 逐行读取config文件 # 返回一个2-D array def read_config(): return_list = [] with open(config_file, "r") as fd: for line in fd.readlines(): # skip empty line if not line.strip('\n'): continue (enable, check_host, check_class, check_name, script_name, operator, threshold, check_args) = \ line.strip('\n').split(':') # skip disabled options if enable != "ENABLE": continue return_list.append([check_host, check_class, check_name, script_name, operator, threshold, check_args]) return return_list # 逐行读取var配置文件 def read_var(): return_dict = {} with open(var_file, "r") as fd: for line in fd: # skip empty line line_lst = line.strip().split('=') return_dict[line_lst[0]] = line_lst[1].strip('"').strip() return return_dict def run_script(script_name, check_args): script_result = None script_file = script_dir + '/' + script_name print("start run script %s %s" % (script_file, check_args)) if not os.path.exists(script_file): print("错误！\t脚本文件: " + script_file + " 不存在") return None if script_name.endswith(".sh"): # print("sh %s %s %s" % (script_file, base_path, check_args)) script_result = os.popen( "sh %s %s %s" % (script_file, base_path, check_args) ).read().splitlines() elif script_name.endswith(".py"): # print("python %s %s %s" % (script_file, base_path, check_args)) script_result = os.popen( "python %s %s %s" % (script_file, base_path, check_args) ).read().splitlines() else: print("错误！\t脚本文件:" + script_file + "格式不支持，请检查扩展名") return script_result def run_script_remote(script_name, check_args, cls_name='', run_host=''): """通过tiup在远程服务器执行""" script_result = None script_file = script_dir + '/' + script_name print("start run script %s %s %s %s" % (script_file, check_args, cls_name, run_host)) if run_host == '': print("错误! \t无有效机器名") return None if not os.path.exists(script_file): print("错误！\t脚本文件: " + script_file + " 不存在") return None if script_name.endswith(".sh"): # print("sh %s %s %s %s" % (script_file, base_path, check_args)) script_result = os.popen( """tiup cluster exec %s --command="sh %s" -N%s""" % (cls_name, script_file, run_host) ).read().splitlines() elif script_name.endswith(".py"): # print("python %s %s %s %s" % (script_file, base_path, check_args)) script_result = os.popen( """tiup cluster exec %s --command="python %s" -N%s""" % (cls_name, script_file, run_host) ).read().splitlines() else: print("错误！\t脚本文件:" + script_file + "格式不支持，请检查扩展名") return script_result # 判断脚本结果是否符合预期 # 只支持可转为float的threshold与script_result # 等于支持字符比较 def compare_threshold(operator, threshold, script_result): if operator == "等于": # return script_result == threshold return str(script_result).strip() == str(threshold) if threshold.endswith('%'): threshold = float(threshold.strip('%')) / 100 elif str(threshold).isdigit(): threshold = float(threshold) if script_result.endswith('%'): script_result = float(script_result.strip('%')) / 100 elif str(script_result).isdigit(): script_result = float(script_result) if operator == "小于等于": return script_result <= threshold elif operator == "大于等于": return script_result >= threshold elif operator == "等于": return script_result == threshold else: print("错误！\t比较方式: " + operator + " 不支持") sys.exit(1) # 判断是否是TiUP管理节点 # 返回tiup或all def host_tiup(): tidb_mng = os.popen('which tiup 2>/dev/null').readlines() if len(tidb_mng) == 0: return "all" else: return "tiup" def handle_result(script_result=[]): """处理脚本返回的结果集""" # 返回数据匹配KV script_result_kv = [] for i_v in script_result: if i_v[:3] == "kv:": script_result_kv.append(i_v[3:]) script_result = script_result_kv result_len = len(script_result) if result_len == 0: check_item = script_name # 当脚本未返回回结果时，使用脚本名作为检查对象 check_value = "无数据" # 当脚本未返回回结果时，检查值未"无数据" if operator == "无数据": check_status = "正常" else: check_status = "异常" # v0.1Beta更新: 出现时间为当前时间 write_log(check_class, check_name, check_item, check_status, check_value, operator, threshold, check_node, current_time) # print_text(check_class, check_name, check_item, check_status, check_value, operator, threshold) return # 当脚本有返回结果时，两行两行地对结果进行审查 # 首先读取的是检查对象，其次是检查值 # 当检查方式为"无数据"时检查结果为"异常" index = 0 # print('000', index, len(script_result)) while index < len(script_result): check_item = script_result[index] check_value = script_result[index + 1] abnormality_key = check_name + check_item # print('000', index, check_item, check_value, abnormality_key) if operator == "无数据": check_status = "异常" elif operator == "大于等于" or operator == "小于等于" or operator == "等于": if compare_threshold(operator, threshold, check_value): check_status = "正常" else: check_status = "异常" else: print("错误！\t比较方式: " + operator + " 不支持") check_status = "脚本错误" write_log(check_class, check_name, check_item, check_status, check_value, operator, threshold, check_node, current_time) # print_text(check_class, check_name, check_item, check_status, check_value, operator, threshold) index += 2 def print_result(logfile=''): """最后一步统一打印日志""" with open(logfile, mode="r") as fd: print_title() for line in fd: if line.startswith("检查类别"): continue (check_class, check_name, check_item, check_status, check_value, operator, check_node, check_time) = \ line.strip('\n').split('||') print_text(check_class, check_name, check_item, check_status, check_value, operator, '', check_node) def run_all(): # 如果没有配置文件，直接退出 if not os.path.exists(config_file): print("错误！\t配置文件:" + config_file + "不存在") sys.exit(1) # 读取要检测的项目及具体的阈值 check_list = read_config() check_var = read_var() # host_falg = host_tiup() # 打印标题，在Log中写入标题行 #print_title() write_log("检查类别", "检查项", "检查对象", "检查结果", "检查值", "检查阈值", "", "检查服务器", "出现时间") global check_host, check_class, check_name, script_name, operator, threshold, check_args, check_node for check_host, check_class, check_name, script_name, operator, threshold, check_args in check_list: # 执行脚本获取结果，这个结果应该为不固定长度的list，偶数列(index = 0, 2, 4)为检查对象，奇数列为对应的检查值 if check_host == "all": host_lst = check_var['CLUSTERNODES'].split(' ') for check_node in host_lst: if check_node == check_var['LOCALNODE']: script_result = run_script(script_name, check_args) else: script_result = run_script_remote(script_name, check_args, check_var['CLUSTERNAME'], check_node) handle_result(script_result) elif check_host == "prometheus": host_lst = check_var['PROMETHEUSNODE'].split(' ') for check_node in host_lst: if check_node == check_var['LOCALNODE']: script_result = run_script(script_name, check_args) else: script_result = run_script_remote(script_name, check_args, check_var['CLUSTERNAME'], check_node) handle_result(script_result) else: check_node = check_var['LOCALNODE'] script_result = run_script(script_name, check_args) handle_result(script_result) time.sleep(5) print_result(output_log) if __name__ == '__main__': run_all()