import os import sqlite3 from concurrent.futures import ThreadPoolExecutor import pandas as pd from datasdk.config import config as sdk_config from diagnosis.dataloader.config import MetricsLoadConfig, SlowLogLoadConfig, EventLoadConfig, SqliteLoadConfig from diagnosis.dataloader.dataloader import MetricsLoader, SlowLogsLoader, EventsLoader, SqliteLoader from .config import * from diagnosis.preprocess.slowlog import config as slowlog_config from diagnosis.preprocess.metrics import config as metrics_config from diagnosis.preprocess.event import config as event_config from diagnosis.dataloader.s3loader import * from diagnosis.preprocess.utils.utils import * from diagnosis.utils.utils import get_sqlite_prefix, build_sqlite, build_temp_sqlite class DefaultTask: config: TaskConfig def __init__(self, config: TaskConfig, loader): print("run default task") self.config = config self.remote_loader = loader def execute(self): sqlite_path = os.path.join(self.config.path, get_sqlite_prefix()) if not os.path.exists(sqlite_path): os.makedirs(sqlite_path) sqlite_path_temp = os.path.join(self.config.path, 'temp/sqlite') if not os.path.exists(sqlite_path_temp): os.makedirs(sqlite_path_temp) metrics_sqlite_name_temp = os.path.join(sqlite_path_temp, self.config.metrics_sqlite_name) metrics_sqlite_name = os.path.join(sqlite_path, self.config.metrics_sqlite_name) slowlog_sqlite_name = os.path.join(sqlite_path, self.config.slowlog_sqlite_name) events_sqlite_name = os.path.join(sqlite_path, self.config.events_sqlite_name) try: # load and preprocess metrics # if not os.path.exists(metrics_sqlite_name): # key = os.path.join(get_sqlite_prefix(), self.config.metrics_sqlite_name) # if self.remote_loader.file_exists(key): # print('{} exist in remote'.format(key)) # file = self.remote_loader.get_file(key) # with open(metrics_sqlite_name, "wb+") as f: # f.write(file) self.metrics_remote_synchronization(sqlite_path, metrics_sqlite_name, metrics_sqlite_name_temp) metrics_load_configs = self.metrics_load_task(metrics_sqlite_name) metrics_preprocess_configs = self.build_metrics_preprocess_configs(metrics_sqlite_name_temp, metrics_load_configs) self.preprocess_task(metrics_preprocess_configs) self.build_metrics_sqlite(sqlite_path_temp, metrics_sqlite_name) self.classify_latency(metrics_sqlite_name, 'latency') # load and preprocess events # events_load_configs = self.events_load_task(events_sqlite_name) # events_preprocess_configs = self.build_events_preprocess_configs(events_sqlite_name, events_load_configs) # self.preprocess_task(events_preprocess_configs) # load and preprocess slowlog if not self.config.skip_slowlog: if not os.path.exists(slowlog_sqlite_name): key = os.path.join(get_sqlite_prefix(), self.config.slowlog_sqlite_name) if self.remote_loader.file_exists(key): print('{} exist in remote'.format(key)) file = self.remote_loader.get_file(key) with open(slowlog_sqlite_name, "wb+") as f: f.write(file) start_ts = self.config.start_ts end_ts = self.config.start_ts while end_ts < self.config.end_ts: end_ts = end_ts + 15 * 60 \ if end_ts + 15 * 60 < self.config.end_ts \ else self.config.end_ts slowlog_load_configs = self.slowlog_load_task(slowlog_sqlite_name, start_ts, end_ts) slowlog_preprocess_configs = self.build_slowlog_preprocess_configs(slowlog_sqlite_name, slowlog_load_configs) self.preprocess_task(slowlog_preprocess_configs) start_ts = end_ts self.convert_slowlog(slowlog_sqlite_name, 'slowlog') else: print("skip to get slowlog") except Exception as e: print("get error {} when preprocess".format(e)) raise else: print("end preprocess default task") for file_name in (self.config.metrics_sqlite_name, self.config.slowlog_sqlite_name): file_path = os.path.join(sqlite_path, file_name) if os.path.exists(file_path): key = os.path.join(get_sqlite_prefix(), file_name) self.remote_loader.upload_file(file_path, key) print("upload sqlite file {}".format(file_path)) def get_metrics_path(self): return os.path.join(self.config.path, "temp/metrics") def get_slowlog_path(self): return os.path.join(self.config.path, "temp/slowlog") def get_events_path(self): return os.path.join(self.config.path, "temp/events") def get_sqlite_path(self): return os.path.join(self.config.path, get_sqlite_prefix()) def metrics_remote_synchronization(self, sqlite_path, metrics_sqlite_name, metrics_sqlite_name_temp): exist = False if not os.path.exists(metrics_sqlite_name): key = os.path.join(get_sqlite_prefix(), self.config.metrics_sqlite_name) if self.remote_loader.file_exists(key): print('{} exist in remote'.format(key)) file = self.remote_loader.get_file(key) with open(metrics_sqlite_name, "wb+") as f: f.write(file) exist = True else: exist = True if exist: print('build temp metrics sqlite') sqlite_loader = SqliteLoader(SqliteLoadConfig(sqlite_path, self.config.metrics_sqlite_name)) data = sqlite_loader.load_data() for name, df in data.items(): build_temp_sqlite(df, metrics_sqlite_name_temp, name) def metrics_load_task(self, sqlite_name): print("start load metric from {} to {}".format(self.config.start_ts, self.config.end_ts)) # check metrics if exists in the remote # key = os.path.join(get_sqlite_prefix(), self.config.metrics_sqlite_name) # if self.remote_loader.file_exists(key): # file = self.remote_loader.get_file(key) # with open(sqlite_name, "wb+") as f: # f.write(file) metrics_load_configs = self._build_metrics_load_configs(sqlite_name) with ThreadPoolExecutor(max_workers=5) as executor: # The order of the exector results is the same as the order of the input, so we can guarantee the chronological order of the returned slowlogs res = executor.map(self.metrics_load_task_one, metrics_load_configs) for one in res: print("{}\n".format(one)) return metrics_load_configs def metrics_load_task_one(self,metrics_load_config): if os.path.exists(metrics_load_config.out_path): return "out path {} already exists, skip load".format(metrics_load_config.out_path) metrics_loader = MetricsLoader( metrics_load_config, sdk_config.Env.prod, self.config.auth0_id, self.config.auth0_secret) metrics_loader.load_data() return "metric %s load done".format(metrics_load_config.rule.name) def _build_metrics_load_configs(self, sqlite_name): configs = [] for r in metrics_rules: expr = r["expr"] # if database_version <= v5.3.1, use another metric for tikv storage if r["name"] == "tikv_store_size_bytes": if 'databaseVersion' in self.config.task_info.keys(): if self.config.task_info['databaseVersion'] <= 'v5.3.1': expr = 'sum(tikv_engine_size_bytes{})' if self.config.skip_tiflash and r["name"].startswith('tiflash'): print("skip to get metrics {}".format(r["name"])) continue start = self.config.start_ts end = self.config.end_ts print("build load config for metric {}".format(r["name"])) p = os.path.join(self.config.path, "temp/metrics") if not os.path.exists(p): os.makedirs(p) # 判断sqlite文件是否存在 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 ='{}'".format( r["name"]) c.execute(sql) tables = c.fetchall() # 如果 table 存在，获取其中 rows 中最大的 timestamp if len(tables) > 0: print("table {} exists".format(r["name"])) sql = "SELECT timestamp FROM {} LIMIT 1 offset (SELECT COUNT(*) - 1 FROM {})".format( r["name"], r["name"]) c.execute(sql) ts = c.fetchall() if len(ts) > 0: max_ts = ts[0][0] if start <= max_ts < end: print("replace start_ts {} to {}".format( start, max_ts)) start = max_ts elif max_ts < start: print("replace start_ts {} to {}".format( start, max_ts)) start = max_ts elif max_ts >= end: print("skip") continue metrics_name = "{}_{}_{}-{}.csv".format( self.config.cluster_id, r["name"], start, end) out = os.path.join(p, metrics_name) # 如果对应的 metrics 文件已经存在，需要将其 remove，以防止其中的数据重复，因为不确定其中的数据是否是完整的 if os.path.exists(out): print("out path already exists, remove") os.remove(out) rule = sdk_config.Rule(r["name"], expr) step = 30 config = MetricsLoadConfig(self.config.cluster_id, rule, start, end, step, out) configs.append(config) return configs def slowlog_load_task(self, sqlite_name, start_ts, end_ts): print("start load slowlog from {} to {}".format(start_ts, end_ts)) slowlog_load_configs = self._build_slowlog_load_configs(sqlite_name, start_ts, end_ts) for c in slowlog_load_configs: slowlog_loader = SlowLogsLoader( c, sdk_config.Env.prod, self.config.auth0_id, self.config.auth0_secret) slowlog_loader.load_data() print("load done\n") return slowlog_load_configs def _build_slowlog_load_configs(self, sqlite_name, start_ts, end_ts): configs = [] print("build load config for slowlog") # start_ts = self.config.start_ts # end_ts = self.config.end_ts p = os.path.join(self.config.path, "temp/slowlog") if not os.path.exists(p): os.makedirs(p) # 判断sqlite文件是否存在 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 configs slowlog_name = "{}_slowlog_{}-{}.csv".format(self.config.cluster_id, start_ts, end_ts) out = os.path.join(p, slowlog_name) if os.path.exists(out): print("out path already exists, remove") os.remove(out) configs.append(SlowLogLoadConfig(self.config.cluster_id, start_ts, end_ts, out)) return configs def events_load_task(self, sqlite_name): print("start load event from {} to {}".format(self.config.start_ts, self.config.end_ts)) # check metrics if exists in the remote if not os.path.exists(sqlite_name): key = os.path.join(get_sqlite_prefix(), self.config.events_sqlite_name) if self.remote_loader.file_exists(key): print('exist in remote') file = self.remote_loader.get_file(key) with open(sqlite_name, "wb+") as f: f.write(file) event_load_configs = self._build_events_load_configs(sqlite_name) for c in event_load_configs: event_loader = EventsLoader(c, self.config.clinic_token) event_loader.load_data() print("load done\n") return event_load_configs def _build_events_load_configs(self, sqlite_name): configs = [] print("build load config for event") start_ts = self.config.start_ts end_ts = self.config.end_ts p = os.path.join(self.config.path, "temp/events") if not os.path.exists(p): os.makedirs(p) # 判断sqlite文件是否存在 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 ='events'" c.execute(sql) tables = c.fetchall() # 如果 table 存在，获取其中 rows 中最大的 timestamp if len(tables) > 0: print("table events exists") sql = "SELECT max(calibration_time) FROM events 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) + 1 elif int(max_ts) < start_ts: print("replace start_ts {} to {}".format(start_ts, max_ts)) start_ts = int(max_ts) + 1 elif int(max_ts) >= end_ts: print("skip") return configs events_name = "{}_events_{}-{}.csv".format(self.config.cluster_id, start_ts, end_ts) out = os.path.join(p, events_name) if os.path.exists(out): print("out path already exists, remove") os.remove(out) configs.append(EventLoadConfig(self.config.cluster_id, start_ts, end_ts, out)) return configs def preprocess_task(self, configs): 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 build_metrics_preprocess_configs(self, sqlite_name_temp, metrics_load_configs): configs = [] i = 0 for c in metrics_load_configs: c1 = metrics_config.ReadFileConfig("read{}".format( i), metrics_read_file_type, [c.out_path], []) configs.append(c1) c2 = metrics_config.CleanDataConfig("clean{}".format(i), metrics_clean_data_type, [], [c1.name], c.start_timestamp, c.end_timestamp, c.step) configs.append(c2) c3 = metrics_config.ConvertMetricsToTupleConfig("convert{}".format(i), metrics_convert_to_tuple_type, [], [c2.name]) configs.append(c3) c4 = metrics_config.CreateMetricsSqliteConfig("sqlite{}".format(i), metrics_create_sqlite_type, [], [c3.name], sqlite_name_temp, c.rule.name) configs.append(c4) i += 1 return configs def build_slowlog_preprocess_configs(self, sqlite_name, slowlog_load_configs): configs = [] for c in slowlog_load_configs: if not os.path.exists(c.out_path): return [] f = open(c.out_path) # f.readline(25) data = f.read() f.close() # os.remove(c.out_path) # split c1 = slowlog_config.SplitConfig( "split", slowlog_split_type, [data], [], '@@\n') configs.append(c1) # parse c2 = slowlog_config.ParseConfig("parse_slowlog", slowlog_parse_type, [], [ c1.name], r"#\s*(.*?)\s*:\s*(.*?)\n") 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) # create slowlog sqlite c9 = slowlog_config.CreateSqliteConfig("create_sqlite", slowlog_create_sqlite_type, [], [c_merge.name], sqlite_name, "slowlog") configs.append(c9) return configs def build_metrics_sqlite(self, sqlite_path_temp, sqlite_path): sqlite_loader = SqliteLoader(SqliteLoadConfig(sqlite_path_temp, self.config.metrics_sqlite_name)) dfs = sqlite_loader.load_data() for name, df in dfs.items(): build_sqlite(df, sqlite_path, name) def build_events_preprocess_configs(self, sqlite_name, event_load_configs): configs = [] for c in event_load_configs: if not os.path.exists(c.out_path): return [] c1 = event_config.ReadFileConfig('read_file', events_read_file_type, [c.out_path], []) configs.append(c1) c2 = event_config.ConvertEventsToTupleConfig('convert_to_tuples', events_convert_to_tuple_type, [], [c1.name]) configs.append(c2) c3 = event_config.CreateEventsSqliteConfig('create_sqlite', events_create_sqlite_type, [], [c2.name], sqlite_name, 'events') configs.append(c3) return configs def convert_slowlog(self, slowlog_sqlite_name, table_name): with sqlite3.connect(slowlog_sqlite_name) as conn: df = pd.read_sql_query( "SELECT * FROM {}".format(table_name), con=conn) if not df.empty: calculate_slowlog_count(df, self.config.start_ts, self.config.end_ts, slowlog_sqlite_name, 30) slowlog_query_time_classification(df, self.config.start_ts, self.config.end_ts, slowlog_sqlite_name, 30) # calculate_slowlog_detail_count(df, self.config.start_ts, self.config.end_ts, slowlog_sqlite_name, 30) def classify_latency(self, metrics_sqlite_name, table_name): with sqlite3.connect(metrics_sqlite_name) as conn: df = pd.read_sql_query( "SELECT * FROM {}".format(table_name), index_col="timestamp", con=conn) if not df.empty: latency_classification(df, metrics_sqlite_name)