import os import json import yaml import time import _thread from typing import Dict from datetime import datetime from diagnosis.dataloader.s3loader import S3Loader from o11ydiagnosis.subtasks.slowlogconverter.objstore import ObjectStore, S3 from o11ydiagnosis.subtasks.logreporter.log import log from o11ydiagnosis.subtasks.logreporter.log_analyzer import process_daily_summary, process_eigenvalue from o11ydiagnosis.subtasks.logreporter.report_rendor import generate_report, update_statistic_csv from o11ydiagnosis.subtasks.logreporter.feishu import send_feishu_message, generate_daily_statistic webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/de223dcc-2873-42d1-8179-3b3c7845bd92" # self-test feishu webhook # webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/e1cde967-15b2-4e97-98da-3acea48f41c5" class Task: s3: ObjectStore cluster: str drhouse_bucket: str drhouse_prefix: str drhouse_local_path: str existed_logdb_paths: Dict[str, str] daily_data_json_file_paths: Dict[str, str] benchmark_data_json_file_paths: Dict[str, str] daily_report_html_file_paths: Dict[str, str] all_statistic_file_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_logdb_paths = {} self.daily_data_json_file_paths = {} self.benchmark_data_json_file_paths = {} self.daily_report_html_file_paths = {} self.all_statistic_file_paths = {} def prepare(self, benchmark_days: int, report_hour_ts: int): utc_hour = datetime.utcfromtimestamp(report_hour_ts).strftime("%Y%m%d%H") log.info(f"[{self.cluster}] calling prepare(report_hour_ts={report_hour_ts},benchmark_days={benchmark_days}) (utc hour: {utc_hour})") assert report_hour_ts % (3600 * 24) == 0 for n in range(24): hour_ts = report_hour_ts + n * 3600 utc_hour = datetime.utcfromtimestamp(hour_ts).strftime("%Y%m%d%H") logdb_s3_key = f"{self.drhouse_prefix}/log-stats/1h/{utc_hour}.db" logdb_local_path = f"{self.drhouse_local_path}/log-stats/1h/{utc_hour}.db" if not os.path.exists(logdb_local_path): log.info(f"[{self.cluster}] {utc_hour} does not exists, downloading from s3") os.makedirs(os.path.dirname(logdb_local_path), exist_ok=True) if self.s3.download(self.drhouse_bucket, logdb_s3_key, logdb_local_path): log.info(f"[{self.cluster}] {utc_hour} downloaded") self.existed_logdb_paths[logdb_local_path] = logdb_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_logdb_paths[logdb_local_path] = logdb_s3_key for n in range(benchmark_days): hour_ts = report_hour_ts - n * 3600 * 24 utc_day = datetime.utcfromtimestamp(hour_ts).strftime("%Y%m%d") report_data_s3_key = f"{self.drhouse_prefix}/log-report/1d/{utc_day}.json" report_data_local_path = f"{self.drhouse_local_path}/log-report/1d/{utc_day}.json" if not os.path.exists(report_data_local_path): log.info(f"[{self.cluster}] {utc_day} does not exists, downloading from s3") os.makedirs(os.path.dirname(report_data_local_path), exist_ok=True) if self.s3.download(self.drhouse_bucket, report_data_s3_key, report_data_local_path): log.info(f"[{self.cluster}] {utc_day} downloaded") self.benchmark_data_json_file_paths[report_data_local_path] = report_data_s3_key else: log.info(f"[{self.cluster}] {utc_day} s3 object does not exists, skipped") else: log.info(f"[{self.cluster}] {utc_day} existed") self.benchmark_data_json_file_paths[report_data_local_path] = report_data_s3_key all_statistic_s3_key = f"{self.drhouse_prefix}/log-report/all/{self.cluster}.csv" all_statistic_local_path = f"{self.drhouse_local_path}/log-report/all/{self.cluster}.csv" if not os.path.exists(all_statistic_local_path): log.info(f"[{self.cluster}] all statistic data does not exists, downloading from s3") os.makedirs(os.path.dirname(all_statistic_local_path), exist_ok=True) if self.s3.download(self.drhouse_bucket, all_statistic_s3_key, all_statistic_local_path): log.info(f"[{self.cluster}] all statistic downloaded") self.all_statistic_file_paths[all_statistic_local_path] = all_statistic_s3_key else: log.info(f"[{self.cluster}] all statistic s3 object does not exists, skipped") else: log.info(f"[{self.cluster}] all statistic existed") self.all_statistic_file_paths[all_statistic_local_path] = all_statistic_s3_key def run(self, benchmark_days: int, report_hour_ts: int): utc_day = datetime.utcfromtimestamp(report_hour_ts).strftime("%Y%m%d") log.info(f"[{self.cluster}] calling run(report_hour_ts={report_hour_ts},benchmark_hours={benchmark_days}) (utc hour: {utc_day})") assert report_hour_ts % (3600 * 24) == 0 # handle daily summary data daily_summary_s3_key = f"{self.drhouse_prefix}/log-report/1d/{utc_day}.json" daily_summary_local_path = f"{self.drhouse_local_path}/log-report/1d/{utc_day}.json" summary = process_daily_summary(self.existed_logdb_paths) summary_with_eigenvalue = process_eigenvalue(self.benchmark_data_json_file_paths, summary) with open(daily_summary_local_path,'w',encoding='utf-8') as f: json.dump(summary_with_eigenvalue, f,ensure_ascii=False) self.daily_data_json_file_paths[daily_summary_local_path] = daily_summary_s3_key # handle html report report_html_s3_key = f"{self.drhouse_prefix}/log-report/1d/{utc_day}.html" report_html_local_path = f"{self.drhouse_local_path}/log-report/1d/{utc_day}.html" generate_report(self.cluster, utc_day, summary_with_eigenvalue, report_html_local_path) # handle feishu message daily_statistic_data = generate_daily_statistic(summary_with_eigenvalue) send_feishu_message(self.cluster, utc_day, daily_statistic_data, self.drhouse_bucket, daily_summary_s3_key, webhook_url) self.daily_report_html_file_paths[report_html_local_path] = report_html_s3_key # handle all statistic csv all_statistic_s3_key = f"{self.drhouse_prefix}/log-report/all/{self.cluster}.csv" all_statistic_local_path = f"{self.drhouse_local_path}/log-report/all/{self.cluster}.csv" update_statistic_csv(utc_day, daily_statistic_data["all_logs_statistics"], all_statistic_local_path) self.all_statistic_file_paths[report_html_local_path] = all_statistic_s3_key def upload_modified_objects(self): for filepath, key in self.daily_data_json_file_paths.items(): self.s3.upload(self.drhouse_bucket, key, filepath) self.daily_data_json_file_paths.clear() for filepath, key in self.daily_report_html_file_paths.items(): self.s3.upload(self.drhouse_bucket, key, filepath) self.daily_report_html_file_paths.clear() for filepath, key in self.all_statistic_file_paths.items(): self.s3.upload(self.drhouse_bucket, key, filepath) self.all_statistic_file_paths.clear() n_thread = 0 def run_cluster(task: Task, ts: int): global n_thread try: task.prepare(7, ts) task.run(7, ts) task.upload_modified_objects() finally: n_thread -= 1 if __name__ == '__main__': s3 = S3() s3loader = S3Loader("o11y-prod-shared-us-east-1") task_list = s3loader.get_file_list("diagnosis/tasks/processor-dedicated/") ts = int(time.time()) ts -= ts % (24*3600) ts -= 24 * 3600 n_thread += len(task_list) for key in task_list: try: task_info_content = s3loader.get_file(key) task_info = yaml.load(task_info_content, yaml.FullLoader) cluster_id = task_info["meta"]["clusterID"] task = Task( s3=s3, cluster=cluster_id, drhouse_bucket="o11y-prod-shared-us-east-1", drhouse_prefix=f"diagnosis/data/{cluster_id}/", drhouse_local_path=f"/data/{cluster_id}/", ) _thread.start_new_thread(run_cluster, (task,ts,)) except Exception: log.exception("failed to run, cluster="+cluster_id) while True: if n_thread > 0: time.sleep(10) continue break