from datasdk.config import config from datasdk.sources import slowlog def getslowlog(auth0Id:str,auth0secret:str,clusterID:str,start:int,end:int,savefile:str,serverless:bool=False): cfg = config.Config(config.Env.prod,auth0Id,auth0secret) slow=slowlog.SlowLog(cfg) data=slow.getSlowLog(clusterID,serverless,start,end) # parse data to sqlite file slowlog_preprocess_configs = build_slowlog_preprocess_configs(savefile, data, clusterID, serverless, ) result=preprocess_task(slowlog_preprocess_configs) def build_slowlog_preprocess_configs(sqlite_name, data, clusterID:str,serverless=False): from diagnosis.preprocess.slowlog import config as slowlog_config from diagnosis.task.preprocess.config import slowlog_split_type,slowlog_parse_type,slowlog_convert_to_dict_type,slowlog_mapping_type,slowlog_merge_map_type,slowlog_table_fields,slowlog_fields,slowlog_create_sqlite_type,slowlog_parse_sql_type,slowlog_analyze_plan_type,slowlog_trace_type configs = [] if data is not None: # split c1 = slowlog_config.SplitConfig( "split", slowlog_split_type, [data], [], '@@\n',serverless) configs.append(c1) # parse c2 = slowlog_config.ParseConfig("parse_slowlog", slowlog_parse_type, [], [ c1.name], r"(\b\w+\b):\s(\b[\w.:'\[\]\()\-\=\+/]+)") configs.append(c2) # convert c3 = slowlog_config.ConvertToDictConfig( "convert_list", slowlog_convert_to_dict_type, [], [c2.name]) configs.append(c3) # mapping c4 = slowlog_config.MappingConfig("mapping1", slowlog_mapping_type, [], [c3.name], slowlog_table_fields, slowlog_fields) configs.append(c4) # parse c5 = slowlog_config.ParseConfig( "parse_base_info", slowlog_parse_type, [], [c1.name], "[^;]+") configs.append(c5) # mapping c6 = slowlog_config.MappingConfig("mapping2", slowlog_mapping_type, [], [c5.name], ["tidb_instance", "ip"], [3, 1]) configs.append(c6) # merge c7 = slowlog_config.MergeConfig( "merge1", slowlog_merge_map_type, [], [c4.name, c6.name]) configs.append(c7) # get sql c_sql = slowlog_config.ParseConfig("get_sql", slowlog_parse_type, [], [c1.name], "^[^#].*;$") configs.append(c_sql) # mapping c_map = slowlog_config.MappingConfig("mapping3", slowlog_mapping_type, [], [c_sql.name], ["sql_statement"], [0]) configs.append(c_map) c_merge = slowlog_config.MergeConfig( "merge2", slowlog_merge_map_type, [], [c7.name, c_map.name]) configs.append(c_merge) # parse sql statement c_parse_sql = slowlog_config.ParseSQLConfig("parse_sql", slowlog_parse_sql_type, [], [c_map.name],"sql_statement") configs.append(c_parse_sql) c_merge_sql = slowlog_config.MergeConfig( "merge3", slowlog_merge_map_type, [], [c_merge.name, c_parse_sql.name]) configs.append(c_merge_sql) # decode sql plan and analyze c_analyze_plan = slowlog_config.AnalyzePlanConfig("analyze_plan", slowlog_analyze_plan_type, [], [c_merge_sql.name],"Plan") configs.append(c_analyze_plan) # merge c_merge_plan = slowlog_config.MergeConfig( "merge4", slowlog_merge_map_type, [], [c_merge_sql.name, c_analyze_plan.name],{"tables":""}) configs.append(c_merge_plan) # fresh db c_fresh_db = slowlog_config.TraceSQLConfig("fresh_db", slowlog_trace_type, [], [c_merge_plan.name],["DB"],"db","Time",clusterID) configs.append(c_fresh_db) c_merge_fresh_db = slowlog_config.MergeConfig( "merge5", slowlog_merge_map_type, [], [c_merge_plan.name, c_fresh_db.name]) configs.append(c_merge_fresh_db) # fresh table c_fresh_table = slowlog_config.TraceSQLConfig("fresh_table", slowlog_trace_type, [], [c_merge_plan.name],["DB","tables"],"table","Time",clusterID) configs.append(c_fresh_table) c_merge_fresh_table = slowlog_config.MergeConfig( "merge6", slowlog_merge_map_type, [], [c_merge_fresh_db.name, c_fresh_table.name]) configs.append(c_merge_fresh_table) # fresh digest c_fresh_digest = slowlog_config.TraceSQLConfig("fresh_digest", slowlog_trace_type, [], [c_merge_plan.name],["Digest"],"sql_digest","Time",clusterID) configs.append(c_fresh_digest) c_merge_fresh_digest = slowlog_config.MergeConfig( "merge7", slowlog_merge_map_type, [], [c_merge_fresh_table.name, c_fresh_digest.name]) configs.append(c_merge_fresh_digest) # fresh partition table; # c_fresh_partition_table = slowlog_config.TraceSQLConfig("fresh_partition_table", slowlog_trace_type, [], # [c_merge_plan.name],"tables","partition_table") # configs.append(c_fresh_partition_table) # c_merge_fresh_partition_table = slowlog_config.MergeConfig( # "merge8", slowlog_merge_map_type, [], [c_merge_fresh_digest.name, c_fresh_partition_table.name]) # configs.append(c_merge_fresh_partition_table) # create slowlog sqlite c9 = slowlog_config.CreateSqliteConfig("create_sqlite", slowlog_create_sqlite_type, [], [c_merge_fresh_digest.name], sqlite_name, "slowlog") configs.append(c9) return configs def preprocess_task(configs): from diagnosis.task.preprocess.config import TypeMap result = {} if len(configs) == 0: return for config in configs: if len(config.data) == 0: if len(config.data_source) == 0: # todo add error continue else: data = [] for ds in config.data_source: data.append(result[ds]) if len(data) == 0: result[config.name] = [] continue processor = TypeMap[config.types](config, data) r = processor.execute() result[config.name] = r else: data = config.data if len(data) == 0: result[config.name] = [] continue processor = TypeMap[config.types](config, data) r = processor.execute() result[config.name] = r return result def continue_slowlog_load(sqlite_name, start_ts, end_ts): import os import sqlite3 print("build load config for slowlog") if os.path.exists(sqlite_name): print("sqlite {} exists".format(sqlite_name)) with sqlite3.connect(sqlite_name) as conn: c = conn.cursor() # 判断 metrics 对应的 table 是否存在 sql = "SELECT name FROM sqlite_master where type ='table' and name ='slowlog'" c.execute(sql) tables = c.fetchall() # 如果 table 存在，获取其中 rows 中最大的 timestamp if len(tables) > 0: print("table slowlog exists") sql = "SELECT max(time) FROM slowlog LIMIT 1" c.execute(sql) ts = c.fetchall() if len(ts) > 0: max_ts = ts[0][0] if start_ts <= int(max_ts) < end_ts: print("replace start_ts {} to {}".format(start_ts, max_ts)) start_ts = int(max_ts) elif int(max_ts) < start_ts: print("replace start_ts {} to {}".format(start_ts, max_ts)) start_ts = int(max_ts) elif int(max_ts) >= end_ts: print("skip") return start_ts, end_ts return start_ts,end_ts