import os import sqlite3 from typing import Dict, Set, Tuple from datetime import datetime from o11ydiagnosis.subtasks.slowlogcompactor.log import log from o11ydiagnosis.subtasks.slowlogconverter.db import StatsDB from o11ydiagnosis.subtasks.slowlogconverter.objstore import ObjectStore from o11ydiagnosis.subtasks.slowlogconverter.slowstats import SlowStats, merge_stats class Task: s3: ObjectStore cluster: str drhouse_bucket: str drhouse_prefix: str drhouse_local_path: str existed_statsdb_paths: Dict[str, str] dirty_merged_statsdb_paths: Dict[str, str] def __init__(self, s3: ObjectStore, cluster: str, drhouse_bucket: str, drhouse_prefix: str, drhouse_local_path: str): self.s3 = s3 self.cluster = cluster self.drhouse_bucket = drhouse_bucket self.drhouse_prefix = drhouse_prefix.rstrip("/") self.drhouse_local_path = drhouse_local_path.rstrip("/") self.existed_statsdb_paths = {} self.dirty_merged_statsdb_paths = {} def prepare(self, begin_hour_ts: int, n_hours: int): utc_hour = datetime.utcfromtimestamp(begin_hour_ts).strftime("%Y%m%d%H") log.info(f"[{self.cluster}] calling prepare(begin_hour_ts={begin_hour_ts},n_hours={n_hours}) (utc hour: {utc_hour})") assert n_hours > 1 assert begin_hour_ts % (n_hours * 3600) == 0 self.existed_statsdb_paths = {} for n in range(n_hours): hour_ts = begin_hour_ts + n * 3600 utc_hour = datetime.utcfromtimestamp(hour_ts).strftime("%Y%m%d%H") statsdb_s3_key = f"{self.drhouse_prefix}/slowlog-stats/1h/{utc_hour}.db" statsdb_local_path = f"{self.drhouse_local_path}/slowlog-stats/1h/{utc_hour}.db" if not os.path.exists(statsdb_local_path): log.info(f"[{self.cluster}] {utc_hour} does not exists, downloading from s3") os.makedirs(os.path.dirname(statsdb_local_path), exist_ok=True) if self.s3.download(self.drhouse_bucket, statsdb_s3_key, statsdb_local_path): log.info(f"[{self.cluster}] {utc_hour} downloaded") self.existed_statsdb_paths[statsdb_local_path] = statsdb_s3_key else: log.info(f"[{self.cluster}] {utc_hour} s3 object does not exists, skipped") else: log.info(f"[{self.cluster}] {utc_hour} existed") self.existed_statsdb_paths[statsdb_local_path] = statsdb_s3_key def run(self, begin_hour_ts: int, n_hours: int): utc_hour = datetime.utcfromtimestamp(begin_hour_ts).strftime("%Y%m%d%H") log.info(f"[{self.cluster}] calling run(begin_hour_ts={begin_hour_ts},n_hours={n_hours}) (utc hour: {utc_hour})") assert n_hours > 1 assert begin_hour_ts % (n_hours * 3600) == 0 all_keys: Set[str] = set() all_meta: Dict[str, Tuple[str, bool]] = {} all_stats: Dict[str, SlowStats] = {} all_stats_by_db: Dict[str, Dict[str, SlowStats]] = {} all_count_per_minute: Dict[str, Dict[int, int]] = {} for statsdb_local_path in self.existed_statsdb_paths.keys(): with sqlite3.connect(statsdb_local_path) as statsdb: keys = StatsDB.query_keys(statsdb) meta = StatsDB.query_meta(statsdb) stats = StatsDB.query_stats(statsdb) stats_by_db = StatsDB.query_stats_by_db(statsdb) count_per_minute = StatsDB.query_count_per_minute(statsdb) log.info(f"[{self.cluster}] loading from {statsdb_local_path}") log.info(f"[{self.cluster}] loaded keys: {len(keys)}") log.info(f"[{self.cluster}] loaded meta count: {len(meta)}") log.info(f"[{self.cluster}] loaded stats count: {len(stats)}") log.info(f"[{self.cluster}] loaded stats_by_db count: {len(stats_by_db)}") log.info(f"[{self.cluster}] loaded count_per_minute count: {len(count_per_minute)}") # Merge keys. all_keys = all_keys.union(keys) # Merge meta. all_meta.update(meta) # Merge stats. for digest, s in stats.items(): if digest not in all_stats: all_stats[digest] = s else: merge_stats(all_stats[digest], s) # Merge stats_by_db. for dbname, ss in stats_by_db.items(): if dbname not in all_stats_by_db: all_stats_by_db[dbname] = ss else: for digest, s in ss.items(): if digest not in all_stats_by_db[dbname]: all_stats_by_db[dbname][digest] = s else: merge_stats(all_stats_by_db[dbname][digest], s) # Merge count_per_minute. for digest, values in count_per_minute.items(): if digest not in all_count_per_minute: all_count_per_minute[digest] = values else: for ts, count in values.items(): if ts not in all_count_per_minute[digest]: all_count_per_minute[digest][ts] = 0 all_count_per_minute[digest][ts] += count utc_hour = datetime.utcfromtimestamp(begin_hour_ts).strftime("%Y%m%d%H") merged_statsdb_s3_key = f"{self.drhouse_prefix}/slowlog-stats/{n_hours}h/{utc_hour}.db" merged_statsdb_local_path = f"{self.drhouse_local_path}/slowlog-stats/{n_hours}h/{utc_hour}.db" log.info(f"[{self.cluster}] merged statsdb: {merged_statsdb_local_path}") log.info(f"[{self.cluster}] all keys: {len(all_keys)}") log.info(f"[{self.cluster}] all meta count: {len(all_meta)}") log.info(f"[{self.cluster}] all stats count: {len(all_stats)}") log.info(f"[{self.cluster}] all stats_by_db count: {len(all_stats_by_db)}") log.info(f"[{self.cluster}] all count_per_minute count: {len(all_count_per_minute)}") if not all_keys: log.info(f"[{self.cluster}] no data, will not create merged statsdb") return os.makedirs(os.path.dirname(merged_statsdb_local_path), exist_ok=True) with sqlite3.connect(merged_statsdb_local_path) as merged_statsdb: StatsDB.try_init_tables(merged_statsdb) StatsDB.save_keys(merged_statsdb, all_keys) StatsDB.save_meta(merged_statsdb, all_meta) StatsDB.save_stats(merged_statsdb, all_stats) StatsDB.save_stats_by_db(merged_statsdb, all_stats_by_db) StatsDB.save_count_per_minute(merged_statsdb, all_count_per_minute) merged_statsdb.commit() self.dirty_merged_statsdb_paths[merged_statsdb_local_path] = merged_statsdb_s3_key log.info(f"[{self.cluster}] dirty merged statsdb paths: {self.dirty_merged_statsdb_paths}") def upload_modified_objects(self): for filepath, key in self.dirty_merged_statsdb_paths.items(): self.s3.upload(self.drhouse_bucket, key, filepath) self.dirty_merged_statsdb_paths.clear()