import os import sys import logging import time import yaml import sqlite3 import json import numpy as np from typing import Dict, List from datetime import datetime from concurrent.futures import ThreadPoolExecutor from o11ydiagnosis.serverless.db import SlowStatsDB, StatementsDB from o11ydiagnosis.subtasks.slowlogconverter.slowlog import StreamParser from o11ydiagnosis.subtasks.slowlogconverter.objstore import ObjectStore, S3, GCS, AzureBlob from o11ydiagnosis.subtasks.slowlogconverter.task import find_matched_keys _h = logging.StreamHandler(sys.stdout) _h.setFormatter(logging.Formatter("%(asctime)s [%(name)s] [%(levelname)s] %(message)s")) log = logging.Logger("serverless", logging.DEBUG) log.addHandler(_h) class Task: s3: ObjectStore bucket: str cache_dir: str def __init__(self, s3: ObjectStore, bucket: str, cache_dir: str): self.s3 = s3 self.bucket = bucket self.cache_dir = cache_dir def run(self, hour_ts: int): raise NotImplementedError class MergeTask: s3: ObjectStore bucket: str cache_dir: str def __init__(self, s3: ObjectStore, bucket: str, cache_dir: str): self.s3 = s3 self.bucket = bucket self.cache_dir = cache_dir def run(self, begin_hour_ts: int, n_hours: int): raise NotImplementedError class Checkpoints: s3: ObjectStore bucket: str cache_dir: str S3_KEY = "diagnosis/checkpoints/serverless/checkpoint.yaml" def __init__(self, s3: ObjectStore, bucket: str, cache_dir: str): self.s3 = s3 self.bucket = bucket self.cache_dir = cache_dir def load(self) -> Dict[str, int]: object = self.s3.object(self.bucket, Checkpoints.S3_KEY) if not object: return {} return dict(yaml.safe_load(object)) def save(self, content: Dict[str, int]): filepath = os.path.join(self.cache_dir, "checkpoint.yaml") with open(filepath, "w") as f: yaml.dump(content, f) self.s3.upload(self.bucket, Checkpoints.S3_KEY, filepath) class ServerlessTask: tasks: List[Task] merge_tasks: List[MergeTask] checkpoints: Checkpoints def __init__(self, provider: str, bucket: str, cache_root: str): if provider == "aws": s3 = S3() elif provider == "gcp": s3 = GCS() elif provider == "azure": s3 = AzureBlob() else: raise ValueError("invalid provider: " + provider) cache_dir = os.path.join(cache_root, "serverless") self.tasks = [ SlowStatsTask(s3, bucket, cache_dir), StatementTask(s3, bucket, cache_dir), ] self.merge_tasks = [ SlowStatsMergeTask(s3, bucket, cache_dir), StatementMergeTask(s3, bucket, cache_dir), ] self.checkpoints = Checkpoints(s3, bucket, cache_dir) def start(self): if self.checkpoints: checkpoints = self.checkpoints.load() now = checkpoints.get("checkpoint_time") if now is None: now = int(time.time()) - (7 * 24 * 60 * 60) else: now = int(time.time()) - (7 * 24 * 60 * 60) now -= now % 600 while True: if now > time.time(): time.sleep(60) continue utc_time = datetime.utcfromtimestamp(now).strftime("%Y-%m-%d %H:%M:%S") log.info(f"execute at {now} ({utc_time})") if now % 3600 == 600: hour_ts = now - 600 - 3600 utc_hour = datetime.utcfromtimestamp(hour_ts).strftime("%Y%m%d%H") log.info(f"running on {hour_ts} ({utc_hour})") with ThreadPoolExecutor(len(self.tasks)) as executor: executor.map(_run_task, self.tasks, [hour_ts] * len(self.tasks)) for n_hours in [3, 6, 12, 24, 72, 168]: if (n_hours < 24 and now % (3600 * n_hours) == 600) or (n_hours >= 24 and now % (3600 * 24) == 600): begin_hour_ts = now - 600 - (3600 * n_hours) utc_hour = datetime.utcfromtimestamp(begin_hour_ts).strftime("%Y%m%d%H") log.info(f"merging on {begin_hour_ts} ({utc_hour}) {n_hours}") with ThreadPoolExecutor(len(self.tasks)) as executor: executor.map( _run_merge_task, self.merge_tasks, [begin_hour_ts] * len(self.merge_tasks), [n_hours] * len(self.merge_tasks), ) if self.checkpoints: self.checkpoints.save({"checkpoint_time": now}) now += 600 class SlowStatsTask(Task): def __init__(self, s3: ObjectStore, bucket: str, cache_dir: str): super().__init__(s3, bucket, cache_dir) def run(self, hour_ts: int): begin = hour_ts end = hour_ts + 3600 utc_hour = datetime.utcfromtimestamp(begin).strftime("%Y%m%d%H") local_path = f"{self.cache_dir}/slowlog-stats/1h/{utc_hour}.db" if os.path.exists(local_path): os.remove(local_path) os.makedirs(os.path.dirname(local_path), exist_ok=True) with sqlite3.connect(local_path) as db: db.isolation_level = None SlowStatsDB.try_init_tables(db) prefix = "0/prod-devtier/serverless-all-in-one/logs/slowlog" matched_keys = find_matched_keys(self.s3, self.bucket, prefix, begin, end) log.info(f"matched slowlog objects: {len(matched_keys)}") if not matched_keys: return stats = {} objs = self.s3.objects(self.bucket, matched_keys) for obj in objs: try: parser = StreamParser(obj) while True: try: sl = parser.read_next() except ValueError as e: log.info(e) continue if not sl: break if not sl.time: continue if sl.time < begin: continue if sl.time >= end: break tenant = sl.serverless_tenant_id project = sl.serverless_project_id cluster = sl.serverless_cluster_id if not tenant or not project or not cluster or not sl.digest: continue k = f"{tenant}/{project}/{cluster}/{sl.digest}/{sl.worker_role}/{sl.is_internal}/{sl.is_tiflash}" if k not in stats: stats[k] = { # meta "serverless_tenant_id": tenant, "serverless_project_id": project, "serverless_cluster_id": cluster, "digest": sl.digest, "query": sl.query, "worker_role": sl.worker_role, "is_internal": sl.is_internal, "is_tiflash": sl.is_tiflash, # stats "count": 1, "query_time_sum": sl.query_time or 0, "query_time_max": sl.query_time or 0, "mem_max_sum": sl.mem_max or 0, "mem_max_max": sl.mem_max or 0, "disk_max_sum": sl.disk_max or 0, "disk_max_max": sl.disk_max or 0, "query_time_list": [], } else: stats[k]["count"] += 1 stats[k]["query_time_sum"] += sl.query_time or 0 stats[k]["mem_max_sum"] += sl.mem_max or 0 stats[k]["disk_max_sum"] += sl.disk_max or 0 if sl.query_time and sl.query_time > stats[k]["query_time_max"]: stats[k]["query_time_max"] = sl.query_time if sl.mem_max and sl.mem_max > stats[k]["mem_max_max"]: stats[k]["mem_max_max"] = sl.mem_max if sl.disk_max and sl.disk_max > stats[k]["disk_max_max"]: stats[k]["disk_max_max"] = sl.disk_max if sl.query_time: stats[k]["query_time_list"].append(sl.query_time) finally: obj.close() for s in stats.values(): query_time_list = s.pop("query_time_list") if query_time_list: query_time_list.sort() s["query_time_p90"] = np.percentile(query_time_list, 90) s["query_time_p99"] = np.percentile(query_time_list, 99) else: s["query_time_p90"] = 0 s["query_time_p99"] = 0 log.info(f"saving slow stats count: {len(stats)}") try: db.execute("BEGIN") SlowStatsDB.save_slow_stats(db, stats.values()) except Exception as e: db.rollback() raise e else: db.commit() s3_key = f"diagnosis/data/serverless/slowlog-stats/1h/{utc_hour}.db" self.s3.upload(self.bucket, s3_key, local_path) class StatementTask(Task): def __init__(self, s3: ObjectStore, bucket: str, cache_dir: str): super().__init__(s3, bucket, cache_dir) def run(self, hour_ts: int): begin = hour_ts end = hour_ts + 3600 utc_hour = datetime.utcfromtimestamp(begin).strftime("%Y%m%d%H") local_path = f"{self.cache_dir}/statements/1h/{utc_hour}.db" if os.path.exists(local_path): os.remove(local_path) os.makedirs(os.path.dirname(local_path), exist_ok=True) with sqlite3.connect(local_path) as db: db.isolation_level = None StatementsDB.try_init_tables(db) prefix = "0/prod-devtier/serverless-all-in-one/logs/statement" matched_keys = find_matched_keys(self.s3, self.bucket, prefix, begin, end) log.info(f"matched statementslog objects: {len(matched_keys)}") if not matched_keys: return stmts = {} objs = self.s3.objects(self.bucket, matched_keys) for obj in objs: try: lines = obj.readlines() for line in lines: try: stmt = json.loads(line) except json.decoder.JSONDecodeError: continue should_continue = False for k in [ "begin", "end", "digest", "serverless_tenant_id", "serverless_project_id", "serverless_cluster_id", ]: if k not in stmt: should_continue = True if should_continue: continue if stmt["begin"] >= end: continue if stmt["end"] < begin: continue # [begin, end), [begin, end), ..., [begin, end) # | | # window begin window end # Each window should contain the left one but not the right one. if stmt["begin"] <= end and stmt["end"] > end: # Skip right one. continue tenant = stmt["serverless_tenant_id"] project = stmt["serverless_project_id"] cluster = stmt["serverless_cluster_id"] digest = stmt["digest"] schema_name = stmt["schema_name"] is_internal = stmt["is_internal"] stmt_type = stmt["stmt_type"] k = f"{tenant}/{project}/{cluster}/{digest}/{schema_name}/{is_internal}/{stmt_type}" if k not in stmts: stmts[k] = { # meta "serverless_tenant_id": tenant, "serverless_project_id": project, "serverless_cluster_id": cluster, "digest": digest, "normalized_sql": stmt["normalized_sql"], "schema_name": schema_name, "is_internal": is_internal, "stmt_type": stmt_type, # stats "exec_count": 0, "ru": 0, "rru": 0, "wru": 0, } stmts[k]["exec_count"] += stmt["exec_count"] or 0 stmts[k]["ru"] += stmt["ru"] or 0 stmts[k]["rru"] += stmt["rru"] or 0 stmts[k]["wru"] += stmt["wru"] or 0 finally: obj.close() log.info(f"saving statements count: {len(stmts)}") try: db.execute("BEGIN") StatementsDB.save_statements(db, stmts.values()) except Exception as e: db.rollback() raise e else: db.commit() s3_key = f"diagnosis/data/serverless/statements/1h/{utc_hour}.db" self.s3.upload(self.bucket, s3_key, local_path) class SlowStatsMergeTask(MergeTask): def __init__(self, s3: ObjectStore, bucket: str, cache_dir: str): super().__init__(s3, bucket, cache_dir) def run(self, begin_hour_ts: int, n_hours: int): hour_ts_list = [begin_hour_ts + n * 3600 for n in range(n_hours)] with ThreadPoolExecutor(100) as executor: results = list(executor.map(self.download, hour_ts_list)) all_stats = {} key_to_nhours = {} for local_path in results: if not local_path: continue with sqlite3.connect(local_path) as db: stats = SlowStatsDB.query_slow_stats(db) log.info(f"loaded slow stats count: {len(stats)}") for s in stats: tenant = s["serverless_tenant_id"] project = s["serverless_project_id"] cluster = s["serverless_cluster_id"] digest = s["digest"] worker_role = s["worker_role"] is_internal = s["is_internal"] is_tiflash = s["is_tiflash"] k = f"{tenant}/{project}/{cluster}/{digest}/{worker_role}/{is_internal}/{is_tiflash}" if k not in key_to_nhours: key_to_nhours[k] = 1 else: key_to_nhours[k] += 1 if k not in all_stats: all_stats[k] = s else: all_stats[k]["count"] += s["count"] all_stats[k]["query_time_sum"] += s["query_time_sum"] all_stats[k]["mem_max_sum"] += s["mem_max_sum"] all_stats[k]["disk_max_sum"] += s["disk_max_sum"] if s["query_time_max"] > all_stats[k]["query_time_max"]: all_stats[k]["query_time_max"] = s["query_time_max"] if s["mem_max_max"] > all_stats[k]["mem_max_max"]: all_stats[k]["mem_max_max"] = s["mem_max_max"] if s["disk_max_max"] > all_stats[k]["disk_max_max"]: all_stats[k]["disk_max_max"] = s["disk_max_max"] all_stats[k]["query_time_p90"] += s["query_time_p90"] all_stats[k]["query_time_p99"] += s["query_time_p99"] for k in all_stats.keys(): all_stats[k]["query_time_p90"] /= key_to_nhours[k] all_stats[k]["query_time_p99"] /= key_to_nhours[k] utc_hour = datetime.utcfromtimestamp(begin_hour_ts).strftime("%Y%m%d%H") local_path = f"{self.cache_dir}/slowlog-stats/{n_hours}h/{utc_hour}.db" log.info(f"saving all slow stats count: {len(all_stats)}") os.makedirs(os.path.dirname(local_path), exist_ok=True) with sqlite3.connect(local_path) as db: SlowStatsDB.try_init_tables(db) SlowStatsDB.save_slow_stats(db, all_stats.values()) db.commit() s3_key = f"diagnosis/data/serverless/slowlog-stats/{n_hours}h/{utc_hour}.db" self.s3.upload(self.bucket, s3_key, local_path) def download(self, hour_ts: int) -> str | None: utc_hour = datetime.utcfromtimestamp(hour_ts).strftime("%Y%m%d%H") s3_key = f"diagnosis/data/serverless/slowlog-stats/1h/{utc_hour}.db" local_path = f"{self.cache_dir}/slowlog-stats/1h/{utc_hour}.db" if os.path.exists(local_path): log.info(f"object existed on local: {local_path}") return local_path else: os.makedirs(os.path.dirname(local_path), exist_ok=True) if not self.s3.download(self.bucket, s3_key, local_path): log.info(f"object not found on s3, skipped: {s3_key}") return None log.info(f"object downloaded: {s3_key}") return local_path class StatementMergeTask(MergeTask): def __init__(self, s3: ObjectStore, bucket: str, cache_dir: str): super().__init__(s3, bucket, cache_dir) def run(self, begin_hour_ts: int, n_hours: int): hour_ts_list = [begin_hour_ts + n * 3600 for n in range(n_hours)] with ThreadPoolExecutor(100) as executor: results = list(executor.map(self.download, hour_ts_list)) all_stmts = {} for local_path in results: if not local_path: continue with sqlite3.connect(local_path) as db: stmts = StatementsDB.query_statements(db) log.info(f"loaded statements count: {len(stmts)}") for s in stmts: tenant = s["serverless_tenant_id"] project = s["serverless_project_id"] cluster = s["serverless_cluster_id"] digest = s["digest"] schema_name = s["schema_name"] is_internal = s["is_internal"] stmt_type = s["stmt_type"] k = f"{tenant}/{project}/{cluster}/{digest}/{schema_name}/{is_internal}/{stmt_type}" if k not in all_stmts: all_stmts[k] = { # meta "serverless_tenant_id": tenant, "serverless_project_id": project, "serverless_cluster_id": cluster, "digest": digest, "normalized_sql": s["normalized_sql"], "schema_name": schema_name, "is_internal": is_internal, "stmt_type": stmt_type, # stats "exec_count": 0, "ru": 0, "rru": 0, "wru": 0, } all_stmts[k]["exec_count"] += s["exec_count"] or 0 all_stmts[k]["ru"] += s["ru"] or 0 all_stmts[k]["rru"] += s["rru"] or 0 all_stmts[k]["wru"] += s["wru"] or 0 if n_hours < 72: all_stmts_list = all_stmts.values() else: cluster_to_stmts = {} for s in all_stmts.values(): cluster = s["serverless_cluster_id"] if cluster not in cluster_to_stmts: cluster_to_stmts[cluster] = [] cluster_to_stmts[cluster].append(s) for cluster in cluster_to_stmts.keys(): if len(cluster_to_stmts[cluster]) > 20: cluster_to_stmts[cluster].sort(key=lambda v: v["ru"], reverse=True) evicted_stmts = cluster_to_stmts[cluster][20:] cluster_to_stmts[cluster] = cluster_to_stmts[cluster][:20] others = evicted_stmts.pop() others["digest"] = None others["normalized_sql"] = None others["schema_name"] = None others["is_internal"] = None others["stmt_type"] = None for s in evicted_stmts: others["exec_count"] += s["exec_count"] others["ru"] += s["ru"] others["wru"] += s["wru"] others["rru"] += s["rru"] cluster_to_stmts[cluster].append(others) all_stmts_list = [] for stmts in cluster_to_stmts.values(): all_stmts_list.extend(stmts) utc_hour = datetime.utcfromtimestamp(begin_hour_ts).strftime("%Y%m%d%H") local_path = f"{self.cache_dir}/statements/{n_hours}h/{utc_hour}.db" log.info(f"saving all statements count: {len(all_stmts_list)}") os.makedirs(os.path.dirname(local_path), exist_ok=True) with sqlite3.connect(local_path) as db: StatementsDB.try_init_tables(db) StatementsDB.save_statements(db, all_stmts_list) db.commit() s3_key = f"diagnosis/data/serverless/statements/{n_hours}h/{utc_hour}.db" self.s3.upload(self.bucket, s3_key, local_path) def download(self, hour_ts: int) -> str | None: utc_hour = datetime.utcfromtimestamp(hour_ts).strftime("%Y%m%d%H") s3_key = f"diagnosis/data/serverless/statements/1h/{utc_hour}.db" local_path = f"{self.cache_dir}/statements/1h/{utc_hour}.db" if os.path.exists(local_path): log.info(f"object existed on local: {local_path}") return local_path else: os.makedirs(os.path.dirname(local_path), exist_ok=True) if not self.s3.download(self.bucket, s3_key, local_path): log.info(f"object not found on s3, skipped: {s3_key}") return None log.info(f"object downloaded: {s3_key}") return local_path def _run_task(task: Task, hour_ts: int): try: task.run(hour_ts) except Exception: log.exception("failed to run task") def _run_merge_task(task: MergeTask, begin_hour_ts: int, n_hours: int): try: task.run(begin_hour_ts, n_hours) except Exception: log.exception("failed to run merge task")