from os import path, environ import sys pa = path.abspath(path.join(path.dirname(__file__), '../..')) sys.path.append(pa) import typer from typing_extensions import Annotated from datasdk.config import config from datasdk.sources import slowlog app = typer.Typer(help="Slowlogs CLI tool.") @app.command() def get(clusterid: Annotated[str, typer.Argument(help="The cluster id of tidbcloud cluster")], start: Annotated[int, typer.Argument(help="The start timestemp for slowlogs range")], end: Annotated[int, typer.Argument(help="The end timestemp for slowlogs range")],tmppath: Annotated[str, typer.Option(help="The tmp data for slowlogs txt file")] ="./tmpData",profile: Annotated[str, typer.Option(help="The profile name of AWS profile",)]="", serverless: Annotated[bool, typer.Option(help="Set this options if cluster type is serverless")] = False): """ get slow logs from object storage with clusterID and time ranges. """ print(f"Get {clusterid} slowlogs from {start} to {end}") # set AWS profile env if profile is not empty if profile != "": environ['AWS_PROFILE'] = profile # get o11y auth0 token from env auth0Id= environ['O11Y_AUTH0_CLIENT_ID'] auth0secret = environ['O11Y_AUTH0_CLIENT_SECRET'] # build save path savepath = f"./{clusterid}_slowlog_{start}_{end}.sqlite" start,end=continue_slowlog_load(savepath,start,end) # split start and end timestemp day by day tmp_start = start while tmp_start + 3600 < end: tmp_end = tmp_start + 3600 getslowlog(auth0Id,auth0secret,clusterid, tmp_start, tmp_end,tmppath,savepath,serverless) tmp_start = tmp_end getslowlog(auth0Id,auth0secret,clusterid, tmp_start, end,tmppath,savepath,serverless) def getslowlog(auth0Id:str,auth0secret:str,clusterID:str,start:int,end:int,filepath:str,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,serverless) result=preprocess_task(slowlog_preprocess_configs) def build_slowlog_preprocess_configs(sqlite_name, data,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 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) # 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 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 if __name__ == "__main__": app()