diff --git a/src/gausskernel/dbmind/tools/index_advisor/extract_log.py b/src/gausskernel/dbmind/tools/index_advisor/extract_log.py index a4c379854..3374ffcb3 100644 --- a/src/gausskernel/dbmind/tools/index_advisor/extract_log.py +++ b/src/gausskernel/dbmind/tools/index_advisor/extract_log.py @@ -1,4 +1,3 @@ - import re import os import argparse @@ -36,12 +35,14 @@ def get_workload_template(templates, sqls): def output_valid_sql(sql): - if 'from pg_' in sql.lower() or ' join ' in sql.lower(): + is_quotation_valid = sql.count("'") % 2 + if 'from pg_' in sql.lower() or ' join ' in sql.lower() or is_quotation_valid: return '' if any(tp in sql.lower() for tp in SQL_TYPE[1:]): - return sql if sql.endswith('; ') else sql + ';' + sql = re.sub(r'for\s+update[\s;]*$', '', sql, flags=re.I) + return sql.strip() if sql.endswith('; ') else sql + ';' elif SQL_TYPE[0] in sql.lower() and 'from ' in sql.lower(): - return sql if sql.endswith('; ') else sql + ';' + return sql.strip() if sql.endswith('; ') else sql + ';' return '' @@ -103,9 +104,11 @@ def get_parsed_sql(file, user, database, sql_amount, statement): reverse=True) for item in param_list: if len(item[1].strip()) >= 256: - sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else ('$' + item[0].strip()), "''") + sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else + ('$' + item[0].strip()), "''") else: - sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else ('$' + item[0].strip()), item[1].strip()) + sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else + ('$' + item[0].strip()), item[1].strip()) if output_valid_sql(sql): SQL_AMOUNT += 1 yield output_valid_sql(sql) @@ -132,34 +135,47 @@ def get_start_position(start_time, file_path): return -1 -def extract_sql_from_log(log_path, output_path, database, user, start_time, sql_amount, statement): - templates = {} - files = os.listdir(log_path) - files = sorted(files, key=lambda x: os.path.getctime(os.path.join(log_path, x)), reverse=True) - valid_files = files - time_stamp = int(time.mktime(time.strptime(start_time, '%Y-%m-%d %H:%M:%S'))) - if start_time: - valid_files = [] - for file in files: - if os.path.getmtime(os.path.join(log_path, file)) < time_stamp: - break - valid_files.insert(0, file) +def record_sql(valid_files, args, output_obj): for ind, file in enumerate(valid_files): - if sql_amount and SQL_AMOUNT >= sql_amount: + if args.sql_amount and SQL_AMOUNT >= args.sql_amount: break - file_path = os.path.join(log_path, file) + file_path = os.path.join(args.l, file) if os.path.isfile(file_path) and re.search(r'.log$', file): start_position = 0 if ind == 0: - start_position = get_start_position(start_time, file_path) + start_position = get_start_position(args.start_time, file_path) if start_position == -1: continue with open(file_path, mode='r') as f: f.seek(start_position, 0) - get_workload_template(templates, get_parsed_sql(f, user, database, sql_amount, - statement)) - with open(output_path, 'w') as output_file: - json.dump(templates, output_file) + if isinstance(output_obj, dict): + get_workload_template(output_obj, get_parsed_sql(f, args.U, args.d, + args.sql_amount, + args.statement)) + else: + for sql in get_parsed_sql(f, args.U, args.d, args.sql_amount, args.statement): + output_obj.write(sql + '\n') + + +def extract_sql_from_log(args): + files = os.listdir(args.l) + files = sorted(files, key=lambda x: os.path.getctime(os.path.join(args.l, x)), reverse=True) + valid_files = files + time_stamp = int(time.mktime(time.strptime(args.start_time, '%Y-%m-%d %H:%M:%S'))) + if args.start_time: + valid_files = [] + for file in files: + if os.path.getmtime(os.path.join(args.l, file)) < time_stamp: + break + valid_files.insert(0, file) + if args.json: + templates = {} + record_sql(valid_files, args, templates) + with open(args.f, 'w') as output_file: + json.dump(templates, output_file) + else: + with open(args.f, 'w') as output_file: + record_sql(valid_files, args, output_file) def main(): @@ -172,17 +188,17 @@ def main(): arg_parser.add_argument("--sql_amount", help="The number of sql collected", type=int) arg_parser.add_argument("--statement", action='store_true', help="Extract statement log type", default=False) + arg_parser.add_argument("--json", action='store_true', + help="Whether the workload file format is json", default=False) args = arg_parser.parse_args() if args.start_time: time.strptime(args.start_time, '%Y-%m-%d %H:%M:%S') if args.sql_amount and args.sql_amount <= 0: raise argparse.ArgumentTypeError("%s is an invalid positive int value" % args.sql_amount) - extract_sql_from_log(args.l, args.f, args.d, args.U, args.start_time, args.sql_amount, - args.statement) + extract_sql_from_log(args) if __name__ == '__main__': main() - diff --git a/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload.py b/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload.py index a1725c0a2..be1cda487 100644 --- a/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload.py +++ b/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload.py @@ -107,6 +107,7 @@ class IndexInfo: self.indexname = indexname self.columns = columns self.indexdef = indexdef + self.primary_key = False self.redundant_obj = [] @@ -164,7 +165,10 @@ def filter_low_benefit(pos_list, candidate_indexes, multi_iter_mode, workload): sql_optimzed += 1 - workload[pos].cost_list[cost_list_pos] / workload[pos].cost_list[0] negative_ratio = (index.insert_sql_num + index.delete_sql_num + index.update_sql_num) / \ index.total_sql_num - if sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio: + # filter the candidate indexes that do not meet the conditions of optimization + if sql_optimzed / len(index.positive_pos) < 0.1: + remove_list.append(key) + elif sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio: remove_list.append(key) for item in sorted(remove_list, reverse=True): candidate_indexes.pop(item) @@ -200,6 +204,9 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total, sql_info = {'sqlDetails': []} benefit_types = [index.ineffective_pos, index.positive_pos, index.negative_pos] for category, benefit_type in enumerate(benefit_types): + sql_count = 0 + for item in benefit_type: + sql_count += workload[item].frequency for ind, pos in enumerate(benefit_type): sql_detail = {} sql_template = workload[pos].statement @@ -208,23 +215,26 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total, sql_detail['sqlTemplate'] = sql_template sql_detail['sql'] = workload[pos].statement + sql_detail['sqlCount'] = sql_count if category == 1: - sql_optimzed = (1 - workload[pos].cost_list[cost_list_pos] / - workload[pos].cost_list[0]) - sql_detail['optimized'] = '%.2f' % (sql_optimzed * 100) + sql_optimzed = (workload[pos].cost_list[0] - + workload[pos].cost_list[cost_list_pos]) / \ + workload[pos].cost_list[cost_list_pos] + sql_detail['optimized'] = '%.3f' % sql_optimzed sql_detail['correlationType'] = category sql_info['sqlDetails'].append(sql_detail) - workload_optimized = 1 - index_cost_total[cost_list_pos] / index_cost_total[0] - sql_info['workloadOptimized'] = '%.2f' % (workload_optimized * 100) + workload_optimized = (1 - index_cost_total[cost_list_pos] / index_cost_total[0]) * 100 + sql_info['workloadOptimized'] = '%.2f' % (workload_optimized if workload_optimized > 1 else 1) sql_info['schemaName'] = SCHEMA sql_info['tbName'] = table_name sql_info['columns'] = index.columns sql_info['statement'] = statement sql_info['dmlCount'] = round(index.total_sql_num) - sql_info['selectRatio'] = round(index.select_sql_num / index.total_sql_num, 2) - sql_info['insertRatio'] = round(index.insert_sql_num / index.total_sql_num, 2) - sql_info['deleteRatio'] = round(index.delete_sql_num / index.total_sql_num, 2) - sql_info['updateRatio'] = round(index.update_sql_num / index.total_sql_num, 2) + sql_info['selectRatio'] = round(index.select_sql_num * 100 / index.total_sql_num, 2) + sql_info['insertRatio'] = round(index.insert_sql_num * 100 / index.total_sql_num, 2) + sql_info['deleteRatio'] = round(index.delete_sql_num * 100 / index.total_sql_num, 2) + sql_info['updateRatio'] = round(100 - sql_info['selectRatio'] - sql_info['insertRatio'] + - sql_info['deleteRatio'], 2) display_info['recommendIndexes'].append(sql_info) return display_info @@ -233,13 +243,25 @@ def record_redundant_indexes(cur_table_indexes, redundant_indexes): cur_table_indexes = sorted(cur_table_indexes, key=lambda index_obj: len(index_obj.columns.split(','))) # record redundant indexes + has_restore = [] for pos, index in enumerate(cur_table_indexes[:-1]): is_redundant = False for candidate_index in cur_table_indexes[pos + 1:]: + if 'UNIQUE INDEX' in index.indexdef: + # ensure that UNIQUE INDEX will not become redundant compared to normal index + if 'UNIQUE INDEX' not in candidate_index.indexdef: + continue + # ensure redundant index not is pkey + elif index.primary_key: + if re.match(r'%s' % candidate_index.columns, index.columns): + candidate_index.redundant_obj.append(index) + redundant_indexes.append(candidate_index) + has_restore.append(candidate_index) + continue if re.match(r'%s' % index.columns, candidate_index.columns): is_redundant = True index.redundant_obj.append(candidate_index) - if is_redundant: + if is_redundant and index not in has_restore: redundant_indexes.append(index) @@ -249,8 +271,14 @@ def check_useless_index(tables): if not tables: return whole_indexes, redundant_indexes tables_string = ','.join(["'%s'" % table for table in tables[SCHEMA]]) - sql = "select tablename, indexname, indexdef from pg_indexes where " \ - "schemaname='%s' and tablename in (%s) order by tablename " % (SCHEMA, tables_string) + sql = "SELECT c.relname AS tablename, i.relname AS indexname, " \ + "pg_get_indexdef(i.oid) AS indexdef, p.contype AS pkey from " \ + "pg_index x JOIN pg_class c ON c.oid = x.indrelid JOIN " \ + "pg_class i ON i.oid = x.indexrelid LEFT JOIN pg_namespace n " \ + "ON n.oid = c.relnamespace LEFT JOIN pg_constraint p ON i.oid = p.conindid" \ + "WHERE (c.relkind = ANY (ARRAY['r'::\"char\", 'm'::\"char\"])) AND " \ + "(i.relkind = ANY (ARRAY['i'::\"char\", 'I'::\"char\"])) AND " \ + "n.nspname = '%s' AND c.relname in (%s) order by c.relname;" % (SCHEMA, tables_string) res = run_shell_cmd([sql]).split('\n') if res: @@ -261,9 +289,11 @@ def check_useless_index(tables): elif re.match(r'\(\d+ rows?\)', line): continue elif '|' in line: - table, index, indexdef = [item.strip() for item in line.split('|')] - cur_columns = re.search(r'\((.*)\)', indexdef).group(1) + table, index, indexdef, pkey = [item.strip() for item in line.split('|')] + cur_columns = re.search(r'\(([^\(\)]*)\)', indexdef).group(1) cur_index_obj = IndexInfo(SCHEMA, table, index, cur_columns, indexdef) + if pkey: + cur_index_obj.primary_key = True whole_indexes.append(cur_index_obj) if cur_table_indexes and cur_table_indexes[-1].table != table: record_redundant_indexes(cur_table_indexes, redundant_indexes) @@ -293,30 +323,39 @@ def check_unused_index_workload(whole_indexes, redundant_indexes, workload_index indexes_name = set(index.indexname for index in whole_indexes) unused_index = list(indexes_name.difference(workload_indexes)) remove_list = [] - for pos, index in enumerate(redundant_indexes): - is_redundant = False - for redundant_obj in index.redundant_obj: - if redundant_obj.indexname not in unused_index: - is_redundant = True - if not is_redundant: - remove_list.append(pos) - for item in sorted(remove_list, reverse=True): - redundant_indexes.pop(item) print_header_boundary(" Current workload useless indexes ") if not unused_index: print("No useless index!") detail_info['uselessIndexes'] = [] # useless index + unused_index_columns = dict() for cur_index in unused_index: - if not re.search('_pkey$', cur_index): - for index in whole_indexes: - if cur_index == index.indexname: + for index in whole_indexes: + if cur_index == index.indexname: + unused_index_columns[cur_index] = index.columns + if 'UNIQUE INDEX' not in index.indexdef: statement = "DROP INDEX %s;" % index.indexname print(statement) useless_index = {"schemaName": index.schema, "tbName": index.table, "type": 1, "columns": index.columns, "statement": statement} detail_info['uselessIndexes'].append(useless_index) print_header_boundary(" Redundant indexes ") + # filter redundant index + for pos, index in enumerate(redundant_indexes): + is_redundant = False + for redundant_obj in index.redundant_obj: + # redundant objects are not in the useless index set or + # equal to the column value in the useless index must be redundant index + index_exist = redundant_obj.indexname not in unused_index_columns.keys() or \ + (unused_index_columns.get(redundant_obj.indexname) and + redundant_obj.columns == unused_index_columns[redundant_obj.indexname]) + if index_exist: + is_redundant = True + if not is_redundant: + remove_list.append(pos) + for item in sorted(remove_list, reverse=True): + redundant_indexes.pop(item) + if not redundant_indexes: print("No redundant index!") # redundant index @@ -378,6 +417,7 @@ def get_workload_template(workload): def workload_compression(input_path): compressed_workload = [] + total_num = 0 if JSON_TYPE: with open(input_path, 'r') as file: templates = json.load(file) @@ -389,7 +429,8 @@ def workload_compression(input_path): for sql in elem['samples']: compressed_workload.append(QueryItem(sql.strip('\n'), elem['cnt'] / len(elem['samples']))) - return compressed_workload + total_num += elem['cnt'] + return compressed_workload, total_num # parse the explain plan to get estimated cost by database optimizer @@ -425,6 +466,7 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No found_plan = False hypo_index = False is_computed = False + select_sql_pos = [] with open(sql_file, 'w') as file: if SCHEMA: file.write('SET current_schema = %s;\n' % SCHEMA) @@ -441,10 +483,14 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No file.write("set explain_perf_mode = 'normal'; \n") for ind, query in enumerate(workload): + if 'select ' not in query.statement.lower(): + workload[ind].cost_list.append(0) + else: + file.write('EXPLAIN ' + query.statement + ';\n') + select_sql_pos.append(ind) # record ineffective sql and negative sql for candidate indexes if is_computed: record_ineffective_negative_sql(index_config[0], query, ind) - file.write('EXPLAIN ' + query.statement + ';\n') result = run_shell_sql_cmd(sql_file).split('\n') if os.path.exists(sql_file): @@ -457,15 +503,18 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No if 'QUERY PLAN' in line: found_plan = True if 'ERROR' in line: - workload.pop(i) + if i >= len(select_sql_pos): + raise ValueError("The size of workload is not correct!") + workload[select_sql_pos[i]].cost_list.append(0) + i += 1 if 'hypopg_create_index' in line: hypo_index = True if found_plan and '(cost=' in line: - if i >= len(workload): + if i >= len(select_sql_pos): raise ValueError("The size of workload is not correct!") query_cost = parse_explain_plan(line) - query_cost *= workload[i].frequency - workload[i].cost_list.append(query_cost) + query_cost *= workload[select_sql_pos[i]].frequency + workload[select_sql_pos[i]].cost_list.append(query_cost) total_cost += query_cost found_plan = False i += 1 @@ -481,8 +530,8 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No ori_indexes_name.add(ind1.strip().split(' ')[1]) else: ori_indexes_name.add(ind2) - while i < len(workload): - workload[i].cost_list.append(0) + while i < len(select_sql_pos): + workload[select_sql_pos[i]].cost_list.append(0) i += 1 if index_config: run_shell_cmd(['SELECT hypopg_reset_index();']) @@ -602,13 +651,13 @@ def get_indexable_columns(table_index_dict): return query_indexable_columns -def generate_candidate_indexes(workload, workload_table_name, iterate=False): +def generate_candidate_indexes(workload, workload_table_name): candidate_indexes = [] index_dict = {} for k, query in enumerate(workload): - table_index_dict = query_index_advisor(query.statement, workload_table_name) - if iterate: + if 'select ' in query.statement.lower(): + table_index_dict = query_index_advisor(query.statement, workload_table_name) need_check = False query_indexable_columns = get_indexable_columns(table_index_dict) valid_index_dict = query_index_check(query.statement, query_indexable_columns) @@ -626,22 +675,20 @@ def generate_candidate_indexes(workload, workload_table_name, iterate=False): need_check = False else: break - else: - valid_index_dict = query_index_check(query.statement, table_index_dict) - # filter duplicate indexes - for table in valid_index_dict.keys(): - if table not in index_dict.keys(): - index_dict[table] = {} - for columns in valid_index_dict[table]: - if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD: - break - workload[k].valid_index_list.append(IndexItem(table, columns)) - if not any(re.match(r'%s' % columns, item) for item in index_dict[table]): - column_sql = {columns: [k]} - index_dict[table].update(column_sql) - elif columns in index_dict[table].keys(): - index_dict[table][columns].append(k) + # filter duplicate indexes + for table in valid_index_dict.keys(): + if table not in index_dict.keys(): + index_dict[table] = {} + for columns in valid_index_dict[table]: + if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD: + break + workload[k].valid_index_list.append(IndexItem(table, columns)) + if not any(re.match(r'%s' % columns, item) for item in index_dict[table]): + column_sql = {columns: [k]} + index_dict[table].update(column_sql) + elif columns in index_dict[table].keys(): + index_dict[table][columns].append(k) for table, column_sqls in index_dict.items(): for column, sql in column_sqls.items(): print("table: ", table, "columns: ", column) @@ -787,14 +834,6 @@ def infer_workload_cost(workload, config, atomic_config_total): min_cost = obj.cost_list[num] total_cost += min_cost - # compute the cost for updating indexes - if 'insert' in obj.statement.lower() or 'delete' in obj.statement.lower(): - for index in config: - index_num = get_index_num(index, atomic_config_total) - if index_num == -1: - raise ValueError("The index isn't found for current query!") - if 0 <= index_num < len(workload[ind].cost_list): - total_cost += obj.cost_list[index_num] - obj.cost_list[0] # record ineffective sql and negative sql for candidate indexes if is_computed: record_ineffective_negative_sql(config[-1], obj, ind) @@ -802,12 +841,12 @@ def infer_workload_cost(workload, config, atomic_config_total): def simple_index_advisor(input_path, max_index_num): - workload = workload_compression(input_path) + workload, workload_count = workload_compression(input_path) print_header_boundary(" Generate candidate indexes ") ori_indexes_name = set() workload_table_name = dict() - display_info = {'recommendIndexes': []} - candidate_indexes = generate_candidate_indexes(workload, workload_table_name, True) + display_info = {'workloadCount': workload_count, 'recommendIndexes': []} + candidate_indexes = generate_candidate_indexes(workload, workload_table_name) if len(candidate_indexes) == 0: print("No candidate indexes generated!") estimate_workload_cost_file(workload, ori_indexes_name=ori_indexes_name) @@ -868,12 +907,12 @@ def greedy_determine_opt_config(workload, atomic_config_total, candidate_indexes def complex_index_advisor(input_path): - workload = workload_compression(input_path) + workload, workload_count = workload_compression(input_path) print_header_boundary(" Generate candidate indexes ") ori_indexes_name = set() workload_table_name = dict() - display_info = {'recommendIndexes': []} - candidate_indexes = generate_candidate_indexes(workload, workload_table_name, True) + display_info = {'workloadCount': workload_count, 'recommendIndexes': []} + candidate_indexes = generate_candidate_indexes(workload, workload_table_name) if len(candidate_indexes) == 0: print("No candidate indexes generated!") estimate_workload_cost_file(workload, ori_indexes_name=ori_indexes_name) @@ -957,3 +996,4 @@ def main(): if __name__ == '__main__': main() + diff --git a/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload_driver.py b/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload_driver.py index c67e98baf..b6760139f 100644 --- a/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload_driver.py +++ b/src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload_driver.py @@ -105,6 +105,7 @@ class IndexInfo: self.indexname = indexname self.columns = columns self.indexdef = indexdef + self.primary_key = False self.redundant_obj = [] @@ -168,7 +169,10 @@ def filter_low_benefit(pos_list, candidate_indexes, multi_iter_mode, workload): sql_optimzed += 1 - workload[pos].cost_list[cost_list_pos] / workload[pos].cost_list[0] negative_ratio = (index.insert_sql_num + index.delete_sql_num + index.update_sql_num) / \ index.total_sql_num - if sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio: + # filter the candidate indexes that do not meet the conditions of optimization + if sql_optimzed / len(index.positive_pos) < 0.1: + remove_list.append(key) + elif sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio: remove_list.append(key) for item in sorted(remove_list, reverse=True): candidate_indexes.pop(item) @@ -204,6 +208,9 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total, sql_info = {'sqlDetails': []} benefit_types = [index.ineffective_pos, index.positive_pos, index.negative_pos] for category, benefit_type in enumerate(benefit_types): + sql_count = 0 + for item in benefit_type: + sql_count += workload[item].frequency for ind, pos in enumerate(benefit_type): sql_detail = {} sql_template = workload[pos].statement @@ -211,23 +218,27 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total, sql_template = re.sub(pattern, '?', sql_template) sql_detail['sqlTemplate'] = sql_template - sql_optimzed = (1 - workload[pos].cost_list[cost_list_pos] / workload[pos].cost_list[0]) sql_detail['sql'] = workload[pos].statement + sql_detail['sqlCount'] = sql_count if category == 1: - sql_detail['optimized'] = '%.2f' % (sql_optimzed * 100) + sql_optimzed = (workload[pos].cost_list[0] - + workload[pos].cost_list[cost_list_pos]) / \ + workload[pos].cost_list[cost_list_pos] + sql_detail['optimized'] = '%.3f' % sql_optimzed sql_detail['correlationType'] = category sql_info['sqlDetails'].append(sql_detail) - workload_optimized = 1 - index_cost_total[cost_list_pos] / index_cost_total[0] - sql_info['workloadOptimized'] = '%.2f' % (workload_optimized * 100) + workload_optimized = (1 - index_cost_total[cost_list_pos] / index_cost_total[0]) * 100 + sql_info['workloadOptimized'] = '%.2f' % (workload_optimized if workload_optimized > 1 else 1) sql_info['schemaName'] = SCHEMA sql_info['tbName'] = table_name sql_info['columns'] = index.columns sql_info['statement'] = statement sql_info['dmlCount'] = round(index.total_sql_num) - sql_info['selectRatio'] = round(index.select_sql_num / index.total_sql_num, 2) - sql_info['insertRatio'] = round(index.insert_sql_num / index.total_sql_num, 2) - sql_info['deleteRatio'] = round(index.delete_sql_num / index.total_sql_num, 2) - sql_info['updateRatio'] = round(index.update_sql_num / index.total_sql_num, 2) + sql_info['selectRatio'] = round(index.select_sql_num * 100 / index.total_sql_num, 2) + sql_info['insertRatio'] = round(index.insert_sql_num * 100 / index.total_sql_num, 2) + sql_info['deleteRatio'] = round(index.delete_sql_num * 100 / index.total_sql_num, 2) + sql_info['updateRatio'] = round(100 - sql_info['selectRatio'] - sql_info['insertRatio'] + - sql_info['deleteRatio'], 2) display_info['recommendIndexes'].append(sql_info) return display_info @@ -236,13 +247,25 @@ def record_redundant_indexes(cur_table_indexes, redundant_indexes): cur_table_indexes = sorted(cur_table_indexes, key=lambda index_obj: len(index_obj.columns.split(','))) # record redundant indexes + has_restore = [] for pos, index in enumerate(cur_table_indexes[:-1]): is_redundant = False for candidate_index in cur_table_indexes[pos + 1:]: + if 'UNIQUE INDEX' in index.indexdef: + # ensure that UNIQUE INDEX will not become redundant compared to normal index + if 'UNIQUE INDEX' not in candidate_index.indexdef: + continue + # ensure redundant index not is pkey + elif index.primary_key: + if re.match(r'%s' % candidate_index.columns, index.columns): + candidate_index.redundant_obj.append(index) + redundant_indexes.append(candidate_index) + has_restore.append(candidate_index) + continue if re.match(r'%s' % index.columns, candidate_index.columns): is_redundant = True index.redundant_obj.append(candidate_index) - if is_redundant: + if is_redundant and index not in has_restore: redundant_indexes.append(index) @@ -252,14 +275,23 @@ def check_useless_index(tables, db): if not tables: return whole_indexes, redundant_indexes tables_string = ','.join(["'%s'" % table for table in tables[SCHEMA]]) - sql = "select tablename, indexname, indexdef from pg_indexes where " \ - "schemaname='%s' and tablename in (%s) order by tablename " % (SCHEMA, tables_string) + sql = "SELECT c.relname AS tablename, i.relname AS indexname, " \ + "pg_get_indexdef(i.oid) AS indexdef, p.contype AS pkey from " \ + "pg_index x JOIN pg_class c ON c.oid = x.indrelid JOIN " \ + "pg_class i ON i.oid = x.indexrelid LEFT JOIN pg_namespace n " \ + "ON n.oid = c.relnamespace LEFT JOIN pg_constraint p ON i.oid = p.conindid" \ + "WHERE (c.relkind = ANY (ARRAY['r'::\"char\", 'm'::\"char\"])) AND " \ + "(i.relkind = ANY (ARRAY['i'::\"char\", 'I'::\"char\"])) AND " \ + "n.nspname = '%s' AND c.relname in (%s) order by c.relname;" % (SCHEMA, tables_string) + res = db.execute(sql) if res: cur_table_indexes = list() for item in res: - cur_columns = re.search(r'\((.*)\)', item[2]).group(1) + cur_columns = re.search(r'\(([^\(\)]*)\)', item[2]).group(1) cur_index_obj = IndexInfo(SCHEMA, item[0], item[1], cur_columns, item[2]) + if item[3]: + cur_index_obj.primary_key = True whole_indexes.append(cur_index_obj) if cur_table_indexes and cur_table_indexes[-1].table != item[0]: record_redundant_indexes(cur_table_indexes, redundant_indexes) @@ -292,30 +324,39 @@ def check_unused_index_workload(whole_indexes, redundant_indexes, workload_index indexes_name = set(index.indexname for index in whole_indexes) unused_index = list(indexes_name.difference(workload_indexes)) remove_list = [] - for pos, index in enumerate(redundant_indexes): - is_redundant = False - for redundant_obj in index.redundant_obj: - if redundant_obj.indexname not in unused_index: - is_redundant = True - if not is_redundant: - remove_list.append(pos) - for item in sorted(remove_list, reverse=True): - redundant_indexes.pop(item) print_header_boundary(" Current workload useless indexes ") if not unused_index: print("No useless index!") detail_info['uselessIndexes'] = [] # useless index + unused_index_columns = dict() for cur_index in unused_index: - if not re.search('_pkey$', cur_index): - for index in whole_indexes: - if cur_index == index.indexname: + for index in whole_indexes: + if cur_index == index.indexname: + unused_index_columns[cur_index] = index.columns + if 'UNIQUE INDEX' not in index.indexdef: statement = "DROP INDEX %s;" % index.indexname print(statement) useless_index = {"schemaName": index.schema, "tbName": index.table, "type": 1, "columns": index.columns, "statement": statement} detail_info['uselessIndexes'].append(useless_index) print_header_boundary(" Redundant indexes ") + # filter redundant index + for pos, index in enumerate(redundant_indexes): + is_redundant = False + for redundant_obj in index.redundant_obj: + # redundant objects are not in the useless index set or + # equal to the column value in the useless index must be redundant index + index_exist = redundant_obj.indexname not in unused_index_columns.keys() or \ + (unused_index_columns.get(redundant_obj.indexname) and + redundant_obj.columns == unused_index_columns[redundant_obj.indexname]) + if index_exist: + is_redundant = True + if not is_redundant: + remove_list.append(pos) + for item in sorted(remove_list, reverse=True): + redundant_indexes.pop(item) + if not redundant_indexes: print("No redundant index!") # redundant index @@ -375,6 +416,7 @@ def get_workload_template(workload): def workload_compression(input_path): + total_num = 0 compressed_workload = [] if JSON_TYPE: with open(input_path, 'r') as file: @@ -387,7 +429,8 @@ def workload_compression(input_path): for sql in elem['samples']: compressed_workload.append(QueryItem(sql.strip('\n'), elem['cnt'] / len(elem['samples']))) - return compressed_workload + total_num += elem['cnt'] + return compressed_workload, total_num # parse the explain plan to get estimated cost by database optimizer @@ -447,17 +490,15 @@ def estimate_workload_cost_file(workload, db, index_config=None, ori_indexes_nam # record ineffective sql and negative sql for candidate indexes if is_computed: record_ineffective_negative_sql(index_config[0], query, ind) - res = db.execute('EXPLAIN ' + query.statement) - if res: - query_cost = parse_explain_plan(res, index_config, ori_indexes_name) - query_cost *= workload[ind].frequency - workload[ind].cost_list.append(query_cost) - total_cost += query_cost + if 'select ' not in query.statement.lower(): + workload[ind].cost_list.append(0) else: - remove_list.append(ind) - - for item in sorted(remove_list, reverse=True): - workload.pop(item) + res = db.execute('EXPLAIN ' + query.statement) + if res: + query_cost = parse_explain_plan(res, index_config, ori_indexes_name) + query_cost *= workload[ind].frequency + workload[ind].cost_list.append(query_cost) + total_cost += query_cost if index_config: db.execute('SELECT hypopg_reset_index()') return total_cost @@ -572,13 +613,13 @@ def get_indexable_columns(table_index_dict): return query_indexable_columns -def generate_candidate_indexes(workload, workload_table_name, db, iterate=False): +def generate_candidate_indexes(workload, workload_table_name, db): candidate_indexes = [] index_dict = {} db.init_conn_handle() for k, query in enumerate(workload): - table_index_dict = query_index_advisor(query.statement, workload_table_name, db) - if iterate: + if 'select ' in query.statement.lower(): + table_index_dict = query_index_advisor(query.statement, workload_table_name, db) need_check = False query_indexable_columns = get_indexable_columns(table_index_dict) valid_index_dict = query_index_check(query.statement, query_indexable_columns, db) @@ -596,22 +637,20 @@ def generate_candidate_indexes(workload, workload_table_name, db, iterate=False) need_check = False else: break - else: - valid_index_dict = query_index_check(query.statement, table_index_dict, db) - # filter duplicate indexes - for table in valid_index_dict.keys(): - if table not in index_dict.keys(): - index_dict[table] = {} - for columns in valid_index_dict[table]: - if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD: - break - workload[k].valid_index_list.append(IndexItem(table, columns)) - if not any(re.match(r'%s' % columns, item) for item in index_dict[table]): - column_sql = {columns: [k]} - index_dict[table].update(column_sql) - elif columns in index_dict[table].keys(): - index_dict[table][columns].append(k) + # filter duplicate indexes + for table in valid_index_dict.keys(): + if table not in index_dict.keys(): + index_dict[table] = {} + for columns in valid_index_dict[table]: + if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD: + break + workload[k].valid_index_list.append(IndexItem(table, columns)) + if not any(re.match(r'%s' % columns, item) for item in index_dict[table]): + column_sql = {columns: [k]} + index_dict[table].update(column_sql) + elif columns in index_dict[table].keys(): + index_dict[table][columns].append(k) for table, column_sqls in index_dict.items(): for column, sql in column_sqls.items(): print("table: ", table, "columns: ", column) @@ -757,14 +796,6 @@ def infer_workload_cost(workload, config, atomic_config_total): min_cost = obj.cost_list[num] total_cost += min_cost - # compute the cost for updating indexes - if 'insert' in obj.statement.lower() or 'delete' in obj.statement.lower(): - for index in config: - index_num = get_index_num(index, atomic_config_total) - if index_num == -1: - raise ValueError("The index isn't found for current query!") - if 0 <= index_num < len(workload[ind].cost_list): - total_cost += obj.cost_list[index_num] - obj.cost_list[0] # record ineffective sql and negative sql for candidate indexes if is_computed: record_ineffective_negative_sql(config[-1], obj, ind) @@ -772,12 +803,12 @@ def infer_workload_cost(workload, config, atomic_config_total): def simple_index_advisor(input_path, max_index_num, db): - workload = workload_compression(input_path) + workload, workload_count = workload_compression(input_path) print_header_boundary(" Generate candidate indexes ") ori_indexes_name = set() workload_table_name = dict() - display_info = {'recommendIndexes': []} - candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db, True) + display_info = {'workloadCount': workload_count, 'recommendIndexes': []} + candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db) db.init_conn_handle() if len(candidate_indexes) == 0: print("No candidate indexes generated!") @@ -840,12 +871,12 @@ def greedy_determine_opt_config(workload, atomic_config_total, candidate_indexes def complex_index_advisor(input_path, db): - workload = workload_compression(input_path) + workload, workload_count = workload_compression(input_path) print_header_boundary(" Generate candidate indexes ") ori_indexes_name = set() workload_table_name = dict() - display_info = {'recommendIndexes': []} - candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db, True) + display_info = {'workloadCount': workload_count, 'recommendIndexes': []} + candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db) db.init_conn_handle() if len(candidate_indexes) == 0: print("No candidate indexes generated!") @@ -932,3 +963,4 @@ def main(): if __name__ == '__main__': main() + diff --git a/src/gausskernel/dbmind/tools/index_advisor/index_server.py b/src/gausskernel/dbmind/tools/index_advisor/index_server.py index ac63d6492..cce849e5d 100644 --- a/src/gausskernel/dbmind/tools/index_advisor/index_server.py +++ b/src/gausskernel/dbmind/tools/index_advisor/index_server.py @@ -53,15 +53,15 @@ class CreateLogger: self.level = level self.log_name = log_name - def create_log(self): + def create_log(self, log_path): logger = logging.getLogger(self.log_name) - log_path = os.path.join(current_dirname, 'log') + log_path = os.path.join(os.path.dirname(log_path), 'log') if os.path.exists(log_path): if os.path.isfile(log_path): os.remove(log_path) os.mkdir(log_path) else: - os.mkdir(log_path) + os.makedirs(log_path) agent_handler = handlers.RotatingFileHandler(filename=os.path.join(log_path, self.log_name), maxBytes=1024 * 1024 * 100, backupCount=5) @@ -74,30 +74,11 @@ class CreateLogger: class IndexServer: - def __init__(self, pid_file, logger, password): + def __init__(self, pid_file, logger, password, **kwargs): self.pid_file = pid_file self.logger = logger - self.app_name = None - self.database = None - self.port = None - self.host = None - self.user = None self.password = password - self.wl_user = None - self.schema = None - self.max_index_num = 0 - self.max_index_storage = 0 - self.driver = False - self.index_intervals = 0 - self.sql_amount = 0 - self.max_generate_log_size = 0 - self.statement = False - self.output_sql_file = None - self.pg_log_path = None - self.datanode = None - self.ai_monitor_url = None - self.log_min_duration_statement = None - self.log_statement = None + self._kwargs = kwargs def check_proc_exist(self, proc_name): """ @@ -123,14 +104,14 @@ class IndexServer: proc = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True) std, err_msg = proc.communicate() if proc.returncode != 0: - self.logger.error("Failed to execute command: %s, \nError: %s." % (cmd, str(err_msg))) + self.logger.error("Failed to execute command. Error: %s." % str(err_msg)) return proc.returncode, std.decode() def save_recommendation_infos(self, recommendation_infos): headers = {'Content-Type': 'application/json'} data = json.dumps(recommendation_infos, default=lambda o: o.__dict__, sort_keys=True, indent=4).encode() - request = Request(url=self.ai_monitor_url, headers=headers, + request = Request(url=self._kwargs['ai_monitor_url'], headers=headers, data=data) response = None @@ -151,31 +132,31 @@ class IndexServer: detail_info_pos = pos + 1 break detail_info_json = json.loads('\n'.join(index_info[detail_info_pos:])) - detail_info_json['appName'] = self.app_name - detail_info_json['nodeHost'] = self.host - detail_info_json['dbName'] = self.database + detail_info_json['appName'] = self._kwargs.get('app_name') + detail_info_json['nodeHost'] = self._kwargs.get('host') + detail_info_json['dbName'] = self._kwargs.get('database') return detail_info_json def execute_index_advisor(self): self.logger.info('Index advisor task starting.') try: - cmd = 'echo %s | python3 %s/index_advisor_workload.py %s %s %s -U %s --h %s -W ' \ + cmd = 'echo %s | python3 %s/index_advisor_workload.py %s %s %s -U %s -W ' \ '--schema %s --json --multi_iter_mode --show_detail' % ( - self.password, current_dirname, self.port, self.database, - self.output_sql_file, self.user, self.host, self.schema) - if self.max_index_storage: - cmd += ' --max_index_storage %s ' % self.max_index_storage - if self.max_index_num: - cmd += ' --max_index_num %s ' % self.max_index_num - if self.driver: + self.password, current_dirname, self._kwargs['port'], self._kwargs['database'], + self._kwargs['output_sql_file'], self._kwargs['user'], self._kwargs['schema']) + if self._kwargs['max_index_storage']: + cmd += ' --max_index_storage %s ' % self._kwargs['max_index_storage'] + if self._kwargs['max_index_num']: + cmd += ' --max_index_num %s ' % self._kwargs['max_index_num'] + if self._kwargs['driver']: try: import psycopg2 cmd = cmd.replace('index_advisor_workload.py', 'index_advisor_workload_driver.py') except ImportError: self.logger.warning('Driver import failed, use gsql to connect to the database.') - self.logger.info('Index advisor cmd:%s' % cmd) - if os.path.exists(self.output_sql_file): + self.logger.info('Index advisor cmd:%s' % cmd.split('|')[-1]) + if os.path.exists(self._kwargs['output_sql_file']): _, res = self.execute_cmd(cmd) detail_info_json = self.convert_output_to_recommendation_infos(res) @@ -188,21 +169,70 @@ class IndexServer: except Exception as e: self.logger.error(e) + def extract_log(self, start_time): + extract_log_cmd = 'python3 %s %s %s --start_time "%s"' % \ + (os.path.join(current_dirname, 'extract_log.py'), + self._kwargs['pg_log_path'], + self._kwargs['output_sql_file'], start_time) + if self._kwargs['database']: + extract_log_cmd += ' -d %s ' % self._kwargs['database'] + if self._kwargs['wl_user']: + extract_log_cmd += ' -U %s ' % self._kwargs['wl_user'] + if self._kwargs['sql_amount']: + extract_log_cmd += ' --sql_amount %s ' % self._kwargs['sql_amount'] + if self._kwargs['statement']: + extract_log_cmd += ' --statement ' + self.logger.info('Extracting log cmd: %s' % extract_log_cmd) + self.execute_cmd(extract_log_cmd) + self.logger.info('The current log extraction is complete.') + + def monitor_log_size(self, guc_reset): + self.logger.info('Open GUC params.') + # get original all file size + original_total_size = self.get_directory_size() + self.logger.info('Original total file size: %sM' % (original_total_size / 1024 / 1024)) + deviation_size = 0 + # open guc + guc_reload = 'gs_guc reload -Z datanode -D {datanode} -c "log_min_duration_statement = 0" && ' \ + 'gs_guc reload -Z datanode -D {datanode} -c "log_statement= \'all\'"' \ + .format(datanode=self._kwargs['datanode']) + self.execute_cmd(guc_reload) + start_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(int(time.time()))) + # caculate log size + count = 0 + while deviation_size < self._kwargs['max_generate_log_size']: + time.sleep(5) + current_size = self.get_directory_size() + deviation_size = (current_size - original_total_size) / 1024 / 1024 + if current_size - original_total_size < 0: + if count >= 60: + break + count += 1 + self.logger.info('Current log size difference: %sM' % deviation_size) + self.logger.info('Start to reset GUC, cmd: %s' % guc_reset) + returncode, res = self.execute_cmd(guc_reset) + if returncode == 0: + self.logger.info('Success to reset GUC setting.') + else: + self.logger.error('Failed to reset GUC params. please check it.') + return start_time + def get_directory_size(self): - files = os.listdir(self.pg_log_path) + files = os.listdir(self._kwargs['pg_log_path']) total_size = 0 for file in files: - total_size += os.path.getsize(os.path.join(self.pg_log_path, file)) + total_size += os.path.getsize(os.path.join(self._kwargs['pg_log_path'], file)) return total_size - def execute_log_index_advisor(self): + def execute_index_recommendation(self): self.logger.info('Start checking guc.') try: guc_check = 'gs_guc check -Z datanode -D {datanode} -c "log_min_duration_statement" && ' \ - 'gs_guc check -Z datanode -D {datanode} -c "log_statement" '.format(datanode=self.datanode) + 'gs_guc check -Z datanode -D {datanode} -c "log_statement" '\ + .format(datanode=self._kwargs['datanode']) returncode, res = self.execute_cmd(guc_check) - origin_min_duration = self.log_min_duration_statement - origin_log_statement = self.log_statement + origin_min_duration = self._kwargs['log_min_duration_statement'] + origin_log_statement = self._kwargs['log_statement'] if returncode == 0: self.logger.info('Original GUC settings is: %s' % res) match_res = re.findall(r'log_min_duration_statement=(\'?[a-zA-Z0-9]+\'?)', res) @@ -220,99 +250,33 @@ class IndexServer: self.logger.info('Test reseting GUC command...') guc_reset = 'gs_guc reload -Z datanode -D %s -c "log_min_duration_statement = %s" && ' \ 'gs_guc reload -Z datanode -D %s -c "log_statement= %s"' % \ - (self.datanode, origin_min_duration, self.datanode, origin_log_statement) + (self._kwargs['datanode'], origin_min_duration, + self._kwargs['datanode'], origin_log_statement) returncode, res = self.execute_cmd(guc_reset) if returncode != 0: guc_reset = 'gs_guc reload -Z datanode -D %s -c "log_min_duration_statement = %s" && ' \ 'gs_guc reload -Z datanode -D %s -c "log_statement= %s"' % \ - (self.datanode, self.log_min_duration_statement, self.datanode, self.log_statement) + (self._kwargs['datanode'], self._kwargs['log_min_duration_statement'], + self._kwargs['datanode'], self._kwargs['log_statement']) ret, res = self.execute_cmd(guc_reset) if ret != 0: raise Exception('Cannot reset GUC initial value, please check it.') self.logger.info('Test successfully') - - self.logger.info('Open GUC params.') - # get original all file size - original_total_size = self.get_directory_size() - self.logger.info('Original total file size: %sM' % (original_total_size/1024/1024)) - deviation_size = 0 - # open guc - guc_reload = 'gs_guc reload -Z datanode -D {datanode} -c "log_min_duration_statement = 0" && ' \ - 'gs_guc reload -Z datanode -D {datanode} -c "log_statement= \'all\'"'.format(datanode=self.datanode) - self.execute_cmd(guc_reload) - start_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(int(time.time()))) - # caculate log size - count = 0 - while deviation_size < self.max_generate_log_size: - time.sleep(5) - current_size = self.get_directory_size() - deviation_size = (current_size - original_total_size)/1024/1024 - if current_size - original_total_size < 0: - if count >= 60: - break - count += 1 - self.logger.info('Current log size difference: %sM' % deviation_size) - self.logger.info('Start to reset GUC, cmd: %s' % guc_reset) - returncode, res = self.execute_cmd(guc_reset) - if returncode == 0: - self.logger.info('Success to reset GUC setting.') - else: - self.logger.error('Failed to reset GUC params. please check it.') - + # open guc and monitor log real-time size + start_time = self.monitor_log_size(guc_reset) # extract log - extract_log_cmd = 'python3 %s %s %s -d %s -U %s --start_time "%s"' % \ - (os.path.join(current_dirname, 'extract_log.py'), - self.pg_log_path, self.output_sql_file, - self.database, self.wl_user, start_time) - if self.sql_amount: - extract_log_cmd += ' --sql_amount %s ' % self.sql_amount - if self.statement: - extract_log_cmd += ' --statement ' - self.logger.info('Extracting log cmd: %s' % extract_log_cmd) - self.execute_cmd(extract_log_cmd) - self.logger.info('The current log extraction is complete.') - - # index adviss + self.extract_log(start_time) + # index advise self.execute_index_advisor() except Exception as e: self.logger.error(e) guc_reset = 'gs_guc reload -Z datanode -D %s -c "log_min_duration_statement = %s" && ' \ 'gs_guc reload -Z datanode -D %s -c "log_statement= %s"' % \ - (self.datanode, self.log_min_duration_statement, self.datanode, - self.log_statement) + (self._kwargs['datanode'], self._kwargs['log_min_duration_statement'], + self._kwargs['datanode'], self._kwargs['log_statement']) self.execute_cmd(guc_reset) - def parse_check_conf(self, config_path): - config = ConfigParser() - config.read(config_path) - self.app_name = config.get("server", "app_name") - self.database = config.get("server", "database") - self.port = config.get("server", "port") - self.host = config.get("server", "host") - self.user = config.get("server", "user") - self.wl_user = config.get("server", "workload_user") - self.schema = config.get("server", "schema") - self.max_index_num = config.getint("server", "max_index_num") - self.max_index_storage = config.get("server", "max_index_storage") - self.driver = config.getboolean("server", "driver") - self.index_intervals = config.getint("server", "index_intervals") - self.sql_amount = config.getint("server", "sql_amount") - self.output_sql_file = config.get("server", "output_sql_file") - self.datanode = config.get("server", "datanode") - self.pg_log_path = config.get("server", "pg_log_path") - self.ai_monitor_url = config.get("server", "ai_monitor_url") - self.max_generate_log_size = config.getfloat("server", "max_generate_log_size") - self.statement = config.getboolean("server", "statement") - self.log_min_duration_statement = config.get("server", "log_min_duration_statement") - self.log_statement = config.get("server", "log_statement") - if not self.log_min_duration_statement or not re.match(r'[a-zA-Z0-9]+', - self.log_min_duration_statement): - raise ValueError("Please enter a legal value of [log_min_duration_statement]") - legal_log_statement = ['none', 'all', 'ddl', 'mod'] - if self.log_statement not in legal_log_statement: - raise ValueError("Please enter a legal value of [log_statement]") - - def start_service(self, config_path): + def start_service(self): # check service is running or not. if os.path.isfile(self.pid_file): pid = self.check_proc_exist("index_server") @@ -320,9 +284,6 @@ class IndexServer: raise Exception("Error: Process already running, can't start again.") else: os.remove(self.pid_file) - # check config file exists - if not os.path.isfile(config_path): - raise Exception("Config file: %s does not exists." % config_path) # get listen host and port self.logger.info("Start service...") @@ -332,12 +293,12 @@ class IndexServer: with open(self.pid_file, mode='w') as f: f.write(str(os.getpid())) - self.parse_check_conf(config_path) - - self.logger.info("Index advisor execution intervals is: %sh" % self.index_intervals) - index_advisor_thread = RepeatTimer(self.index_intervals*60*60, self.execute_log_index_advisor) + self.logger.info("Index advisor execution intervals is: %sh" % + self._kwargs['index_intervals']) + index_recommendation_thread = RepeatTimer(self._kwargs['index_intervals']*60*60, + self.execute_index_recommendation) self.logger.info("Start timer...") - index_advisor_thread.start() + index_recommendation_thread.start() def read_input_from_pipe(): @@ -356,12 +317,46 @@ def read_input_from_pipe(): return input_str +def parse_check_conf(config_path): + config = ConfigParser() + config.read(config_path) + config_dict = dict() + config_dict['app_name'] = config.get("server", "app_name") + config_dict['database'] = config.get("server", "database") + config_dict['port'] = config.get("server", "port") + config_dict['host'] = config.get("server", "host") + config_dict['user'] = config.get("server", "user") + config_dict['wl_user'] = config.get("server", "workload_user") + config_dict['schema'] = config.get("server", "schema") + config_dict['max_index_num'] = config.getint("server", "max_index_num") + config_dict['max_index_storage'] = config.get("server", "max_index_storage") + config_dict['driver'] = config.getboolean("server", "driver") + config_dict['index_intervals'] = config.getint("server", "index_intervals") + config_dict['sql_amount'] = config.getint("server", "sql_amount") + config_dict['output_sql_file'] = config.get("server", "output_sql_file") + config_dict['datanode'] = config.get("server", "datanode") + config_dict['pg_log_path'] = config.get("server", "pg_log_path") + config_dict['ai_monitor_url'] = config.get("server", "ai_monitor_url") + config_dict['max_generate_log_size'] = config.getfloat("server", "max_generate_log_size") + config_dict['statement'] = config.getboolean("server", "statement") + config_dict['log_min_duration_statement'] = config.get("server", "log_min_duration_statement") + config_dict['log_statement'] = config.get("server", "log_statement") + if not config_dict['log_min_duration_statement'] or \ + not re.match(r'[a-zA-Z0-9]+', config_dict['log_min_duration_statement']): + raise ValueError("Please enter a legal value of [log_min_duration_statement]") + legal_log_statement = ['none', 'all', 'ddl', 'mod'] + if config_dict['log_statement'] not in legal_log_statement: + raise ValueError("Please enter a legal value of [log_statement]") + return config_dict + + def manage_service(): - LOGGER = CreateLogger("debug", "start_service.log").create_log() + config_path = os.path.join(current_dirname, 'database-info.conf') + config_dict = parse_check_conf(config_path) + LOGGER = CreateLogger("debug", "start_service.log").create_log(config_dict.get('output_sql_file')) server_pid_file = os.path.join(current_dirname, 'index_server.pid') password = read_input_from_pipe() - IndexServer(server_pid_file, LOGGER, password).start_service( - os.path.join(current_dirname, 'database-info.conf')) + IndexServer(server_pid_file, LOGGER, password, **config_dict).start_service() def main(): @@ -375,3 +370,4 @@ def main(): if __name__ == '__main__': main() +