import json import os import sqlite3 import time from datetime import datetime, timedelta, timezone import yaml from .analyzer import SlowlogAnalyzer, SlowlogComparator, render_html, render_simple_html from .task import getslowlog class Config: PROVIDER_AWS = 'aws' PROVIDER_GCP = 'gcp' REPORT_INTERVAL_HOURSLY = "hourly" REPORT_INTERVAL_DAILY = "daily" REPORT_INTERVAL_WEEKLY = "weekly" def __init__(self, cluster_id=None, project_id=None, tenant_id=None, cache_path=None, o11y_auth0_client_id=None, o11y_auth0_client_secret=None, bucket_name=None): self.cluster_id = cluster_id if cluster_id is not None else os.environ["CLUSTER_ID"] self.project_id = project_id if project_id is not None else os.environ["PROJECT_ID"] self.tenant_id = tenant_id if tenant_id is not None else os.environ["TENANT_ID"] self.cache_path = cache_path if cache_path is not None else os.environ["CACHE_PATH"] self.o11y_auth0_client_id = o11y_auth0_client_id if o11y_auth0_client_id is not None else os.environ[ "O11Y_AUTH0_CLIENT_ID"] self.o11y_auth0_client_secret = o11y_auth0_client_secret if o11y_auth0_client_secret is not None else \ os.environ[ "O11Y_AUTH0_CLIENT_SECRET"] self.bucket_name = bucket_name if bucket_name is not None else os.environ["BUCKET_NAME"] self.local_dev = os.environ["LOCAL_DEV"] == 'True' if "LOCAL_DEV" in os.environ else False self.start_timestemp = int( os.environ["START_TIMESTEMP"]) if "START_TIMESTEMP" in os.environ else None self.end_timestemp = int( os.environ["END_TIMESTEMP"]) if "END_TIMESTEMP" in os.environ else None self.provider = os.environ["PROVIDER"] if "PROVIDER" in os.environ else Config.PROVIDER_AWS self.gcp_project = os.environ["GCP_PROJECT"] if "GCP_PROJECT" in os.environ else None self.local_load_path = os.environ["LOCAL_LOAD_PATH"] if "LOCAL_LOAD_PATH" in os.environ else None self.serverless = os.environ["SERVERLESS"] == 'True' if "SERVERLESS" in os.environ else False self.report_interval_type = os.environ[ "REPORT_INTERVAL_TYPE"] if "REPORT_INTERVAL_TYPE" in os.environ else self.REPORT_INTERVAL_DAILY def get_checkpoint_prefix(self): return f"slow_query/{self.tenant_id}/{self.project_id}/{self.cluster_id}/slowquery/checkpoints" def get_report_prefix(self, timestr): return f"slow_report/{self.tenant_id}/{self.project_id}/{self.cluster_id}/{self.report_interval_type}/{timestr}" def get_sqlite_megre_prefix(self, timestr): return f"slow_query/{self.tenant_id}/{self.project_id}/{self.cluster_id}/slowquery/sqlite_merge/{self.report_interval_type}/{timestr}" def get_sqlite_prefix(self, timestr): return f"slow_query/{self.tenant_id}/{self.project_id}/{self.cluster_id}/slowquery/sqlite/{timestr}" def get_trace_sqlite_prefix(self): return f"slow_query/{self.tenant_id}/{self.project_id}/{self.cluster_id}/slowquery/sqlite/trace" class Manager: def __init__(self, remoteloader, config: Config) -> None: self.remoteloader = remoteloader self.config = config def do_reportings(self, start_ts, end_ts): now_start = start_ts while True: if now_start >= end_ts: break if self.config.report_interval_type == Config.REPORT_INTERVAL_HOURSLY: start_of_time, start_of_next_time = self.get_time_range(now_start) elif self.config.report_interval_type == Config.REPORT_INTERVAL_DAILY: start_of_time, start_of_next_time = self.get_day_range(now_start) elif self.config.report_interval_type == Config.REPORT_INTERVAL_WEEKLY: start_of_time, start_of_next_time = self.get_week_range(now_start) if self.check_report_exists(start_of_time): now_start = start_of_next_time continue report_files, merged_db = self.analyzeSlowlog(start_of_time) for rf in report_files: self.upload_report(rf, start_of_time) print(f"reporting {start_of_time} done") if merged_db is not None: self.upload_merge_sqlite(merged_db, start_of_time) print(f"uploading merged db {merged_db} done") def mapping_hour_report(self, start_of_hour, start_of_next_hour): if self.config.report_interval_type == Config.REPORT_INTERVAL_HOURSLY: # we do not need to analyze slowlog if it is hourly self.do_reportings(start_of_hour, start_of_next_hour) elif self.config.report_interval_type == Config.REPORT_INTERVAL_DAILY: # we do not need to analyze slowlog if it is daily if self.get_day_different(start_of_hour, start_of_next_hour): start_time, start_of_next_time = self.get_day_range(start_of_hour) self.do_reportings(start_time, start_of_next_time) elif self.config.report_interval_type == Config.REPORT_INTERVAL_WEEKLY: # we do not need to analyze slowlog if it is weekly if self.get_week_different(start_of_hour, start_of_next_hour): start_time, start_of_next_time = self.get_week_range(start_of_hour) self.do_reportings(start_time, start_of_next_time) def start(self): # get checkpoint if start from start_timestemp self.load_trace_sqlite() now_start = self.load_checkpoint() if now_start is None: now_start = self.config.start_timestemp else: # if not None,then we should doing reporting from start_timestemp to now_start self.do_reportings(self.config.start_timestemp, now_start) while True: # if end_timestemp is not None and now_start > end_timestemp: if self.config.end_timestemp is not None and now_start >= self.config.end_timestemp: break # we deal time range by left close and right open start_of_hour, start_of_next_hour = self.get_time_range(now_start) # check one hour sqlite exits，if exits, continue next hour if self.check_sqlite_exists(start_of_hour, start_of_next_hour): self.mapping_hour_report(start_of_hour, start_of_next_hour) self.save_checkpoint(start_of_next_hour) now_start = start_of_next_hour continue else: # get now timestamp,if now > start_of_next_hour,then get log range from start_of_hour to start_of_next_hour,and continue next hour now = int(datetime.utcnow().timestamp()) if now > start_of_next_hour: filepath = self.get_slowlog(start_of_hour, start_of_next_hour) # we need update checkpoint and upload sqlite to remote self.upload_trace_sqlite() self.upload_sqlite(filepath, start_of_hour) self.mapping_hour_report(start_of_hour, start_of_next_hour) self.save_checkpoint(start_of_next_hour) now_start = start_of_next_hour continue else: # if now < start_of_next_hour,then sleep and wait max 5 minutes sleep_time = (start_of_next_hour - now) % 300 print(f"sleep {sleep_time} seconds") time.sleep(sleep_time) def load_checkpoint(self): file_name = "checkpoint.yaml" file_path = os.path.join(self.config.get_checkpoint_prefix(), file_name) # check if checkpoint not exists in remote , then return None if not self.remoteloader.file_exists(file_path): return None else: f = self.remoteloader.get_file(file_path) checkpoint = yaml.safe_load(f) return int(checkpoint["checkpoint_time"]) def get_time_range(self, timestamp): dt = datetime.utcfromtimestamp(timestamp).replace(tzinfo=timezone.utc) # get the start of this hour start_of_hour = dt.replace(minute=0, second=0, microsecond=0) # get the start of the next hour start_of_next_hour = start_of_hour + timedelta(hours=1) return int(start_of_hour.timestamp()), int(start_of_next_hour.timestamp()) def get_day_range(self, timestamp): dt = datetime.utcfromtimestamp(timestamp).replace(tzinfo=timezone.utc) start_of_day = dt.replace(hour=0, minute=0, second=0, microsecond=0) start_of_next_day = start_of_day + timedelta(days=1) return int(start_of_day.timestamp()), int(start_of_next_day.timestamp()) def get_week_range(self, timestamp): dt = datetime.utcfromtimestamp(timestamp).replace(tzinfo=timezone.utc) start_of_week = dt - timedelta(days=dt.weekday()) start_of_week = start_of_week.replace(hour=0, minute=0, second=0, microsecond=0) start_of_next_week = start_of_week + timedelta(weeks=1) return int(start_of_week.timestamp()), int(start_of_next_week.timestamp()) def get_day_different(self, start_ts, end_ts): start = datetime.utcfromtimestamp(start_ts).replace(tzinfo=timezone.utc) end = datetime.utcfromtimestamp(end_ts).replace(tzinfo=timezone.utc) return start.date() != end.date() def get_week_different(self, start_ts, end_ts): start = datetime.utcfromtimestamp(start_ts).replace(tzinfo=timezone.utc) end = datetime.utcfromtimestamp(end_ts).replace(tzinfo=timezone.utc) start_week = start.isocalendar()[0:2] end_week = end.isocalendar()[0:2] return start_week != end_week def check_sqlite_exists(self, start_ts, end_ts): file_name = f"{start_ts}_{end_ts}_slowlog.sqlite" # get utc time str timestr = datetime.utcfromtimestamp(start_ts).replace(tzinfo=timezone.utc).strftime('%Y%m%d%H') file_path = os.path.join(self.config.get_sqlite_prefix(timestr), file_name) return self.remoteloader.file_exists(file_path) def check_merge_sqlite_exists(self, start_ts, end_ts): file_name = f"{start_ts}_{end_ts}_slowlog.sqlite" # get utc time str timestr = self.get_report_timestr(start_ts) file_path = os.path.join(self.config.get_sqlite_megre_prefix(timestr), file_name) return self.remoteloader.file_exists(file_path) def get_report_timestr(self, start_ts): timedate = datetime.utcfromtimestamp(start_ts).replace(tzinfo=timezone.utc) if self.config.report_interval_type == Config.REPORT_INTERVAL_HOURSLY: timestr = timedate.strftime('%Y%m%d%H') elif self.config.report_interval_type == Config.REPORT_INTERVAL_DAILY: timestr = timedate.strftime('%Y%m%d') elif self.config.report_interval_type == Config.REPORT_INTERVAL_WEEKLY: timestr = timedate.strftime('%Y%m') + str(self.get_week_of_month(timedate)) return timestr def get_report_file_name(self, start_ts): timestr = self.get_report_timestr(start_ts) json_file = f"{timestr}_slowlog.json" html_file = f"{timestr}_slowlog.html" simple_html_file = f"simple_{timestr}_slowlog.html" return json_file, html_file, simple_html_file def get_week_of_month(self, query_date: datetime): end = int(query_date.strftime("%W")) begin = int(datetime(query_date.year, query_date.month, 1).strftime("%W")) return end - begin + 1 def check_report_exists(self, start_ts): timestr = self.get_report_timestr(start_ts) _, html_file, _ = self.get_report_file_name(start_ts) file_path = os.path.join(self.config.get_report_prefix(timestr), html_file) return self.remoteloader.file_exists(file_path) def get_slowlog(self, start_ts, end_ts): file_name = f"{start_ts}_{end_ts}_slowlog.sqlite" file_path = os.path.join(self.config.cache_path, file_name) if self.check_sqlite_exists(start_ts, end_ts): timestr = datetime.utcfromtimestamp(start_ts).replace(tzinfo=timezone.utc).strftime('%Y%m%d%H') remote_path = os.path.join(self.config.get_sqlite_prefix(timestr), file_name) f = self.remoteloader.get_file(remote_path) with open(file_path, "wb") as cache_file: cache_file.write(f) else: getslowlog( self.config.o11y_auth0_client_id, self.config.o11y_auth0_client_secret, self.config.cluster_id, start_ts, end_ts, file_path, self.config.serverless, ) return file_path def upload_sqlite(self, filepath, start_ts): timestr = datetime.utcfromtimestamp(start_ts).replace(tzinfo=timezone.utc).strftime('%Y%m%d%H') s3_key = os.path.join(self.config.get_sqlite_prefix(timestr), os.path.basename(filepath)) self.remoteloader.upload_file(filepath, s3_key) def upload_report(self, filepath, start_ts): timestr = self.get_report_timestr(start_ts) s3_key = os.path.join(self.config.get_report_prefix(timestr), os.path.basename(filepath)) self.remoteloader.upload_file(filepath, s3_key) def upload_merge_sqlite(self, filepath, start_ts): # get utc time str timestr = self.get_report_timestr(start_ts) s3_key = os.path.join(self.config.get_sqlite_megre_prefix(timestr), os.path.basename(filepath)) self.remoteloader.upload_file(filepath, s3_key) def save_checkpoint(self, timestamp): file_name = "checkpoint.yaml" file_path = os.path.join(self.config.cache_path, file_name) checkpoint = {"checkpoint_time": timestamp} # save to cache path with open(file_path, "w") as f: yaml.dump(checkpoint, f) s3_key = os.path.join(self.config.get_checkpoint_prefix(), file_name) self.remoteloader.upload_file(file_path, s3_key) def upload_trace_sqlite(self): file_name = "trace.sqlite" remote_file_path = os.path.join(self.config.get_trace_sqlite_prefix(), file_name) cache_file_path = os.path.join(self.config.cache_path, file_name) print(remote_file_path, cache_file_path) self.remoteloader.upload_file(cache_file_path, remote_file_path) def load_trace_sqlite(self): file_name = "trace.sqlite" remote_file_path = os.path.join(self.config.get_trace_sqlite_prefix(), file_name) cache_file_path = os.path.join(self.config.cache_path, file_name) if self.remoteloader.file_exists(remote_file_path): f = self.remoteloader.get_file(remote_file_path) with open(cache_file_path, "wb") as cache_file: cache_file.write(f) else: print(cache_file_path) conn = sqlite3.connect(cache_file_path) if not table_exists(conn, "trace"): c = conn.cursor() c.execute('''CREATE TABLE trace (KEY string PRIMARY KEY NOT NULL, FIRST_TIME TIMESTAMP DEFAULT CURRENT_TIMESTAMP);''') conn.commit() conn.close() return def analyzeSlowlog(self, start_ts): step = 3600 # one file per hour json_file, html_file, simple_html_file = self.get_report_file_name(start_ts) json_path = os.path.join(self.config.cache_path, json_file) html_path = os.path.join(self.config.cache_path, html_file) simple_html_path = os.path.join(self.config.cache_path, simple_html_file) i = 0 if self.config.report_interval_type == Config.REPORT_INTERVAL_HOURSLY: max_length = 1 elif self.config.report_interval_type == Config.REPORT_INTERVAL_DAILY: max_length = 24 elif self.config.report_interval_type == Config.REPORT_INTERVAL_WEEKLY: max_length = 168 sqlpath_list = [] prev_sqlpath_list = [] while i < max_length: sqlpath_list.append(self.get_slowlog(start_ts + i * step, start_ts + (i + 1) * step)) prev_sqlpath_list.append(self.get_slowlog(start_ts - (i + 1) * step, start_ts - i * step)) i += 1 print("----- sqlpath_list: {}".format(sqlpath_list)) analyzer_result = SlowlogAnalyzer(sqlpath_list).execute() print("----- prev_sqlpath_list: {}".format(prev_sqlpath_list)) compare_result = SlowlogComparator(sqlpath_list, prev_sqlpath_list).execute() json_dict = { **analyzer_result, **compare_result, } with open(json_path, "w") as f: json.dump(json_dict, f) with open(html_path, "w") as f: f.write(render_html(json_dict)) with open(simple_html_path, "w") as f: f.write(render_simple_html(json_dict)) # merge sqlite merged_db = None if self.config.report_interval_type != Config.REPORT_INTERVAL_HOURSLY: merged_db = self.check_merge_sqlite(sqlpath_list, start_ts, start_ts + (max_length) * step) return [json_path, html_path, simple_html_path], merged_db def check_merge_sqlite(self, sqlite_list, start_ts, end_ts): from .merge import merge_sqlite if self.check_merge_sqlite_exists(start_ts, end_ts): return file_name = f"{start_ts}_{end_ts}_slowlog.sqlite" merged_db = os.path.join(self.config.cache_path, file_name) merge_sqlite(sqlite_list, merged_db, "slowlog") return merged_db def table_exists(conn, table_name): cursor = conn.cursor() query = "SELECT name FROM sqlite_master WHERE type='table' AND name=?" result = cursor.execute(query, (table_name,)).fetchone() cursor.close() return result is not None