import asyncio import datetime import json import time from math import isnan from os import stat from pathlib import Path import numpy as np import pandas as pd from context import ClinicContext from mathbox.app.signal.filter import moving_median from mathbox.calculus.differential import diff from mathbox.statistics.estimator import mean, pearson_correlation_coefficient from utils import cal_imbalance, cal_trend, get_all_metrics, max_key_value, quantile from yaml import Loader, load def search(metrics, mname): results = [metric for metric in metrics if metric.name == mname] if len(results) == 0: return None elif len(results) > 1: # modify return results[0] else: return results[0] def cal_mean_and_trend(all_metrics): avg_stats = {} trend_stats = {} for metrics in all_metrics: # metrics_meta = metrics.signals.describe() # metrics_meta = pd.concat([metrics_meta, metrics.trend()]) try: metrics_des = metrics.signals.describe() except: raise Exception("No data in metrics") if metrics.name in [ "act_conn:total", "connection:total", "qps:total", "avg_cluster_duration:total", ]: metrics_trend = metrics.trend() # too slow so only calculate neccessary variables trend_stats[metrics.name] = metrics_trend.loc["trend"].values[0] if metrics.signals.shape[1] == 1: # avg_stats[metrics.name] = metrics_meta.loc['mean'].values[0] # trend_stats[metrics.name] = metrics_meta.loc['trend'].values[0] if "by_instance" in metrics.name or "by_type" in metrics.name: avg_stats[metrics.name] = { metrics.signals.columns[0]: metrics_des.loc["mean"].values[0] } else: avg_stats[metrics.name] = metrics_des.loc["mean"].values[0] # trend_stats[metrics.name] = metrics_trend.loc['trend'].values[0] elif metrics.signals.shape[1] > 1: avg_stats[metrics.name] = {} # trend_stats[metrics.name] = {} for col in metrics.signals.columns: # avg_stats[metrics.name][col] = metrics_meta.loc['mean'][col] # trend_stats[metrics.name][col] = metrics_meta.loc['trend'][col] avg_stats[metrics.name][col] = metrics_des.loc["mean"][col] # trend_stats[metrics.name][col] = metrics_trend.loc['trend'][col] return avg_stats, trend_stats async def handler(context: ClinicContext): start: int = int(context.start) end: int = int(context.stop) print("job start, perpare_data ...") template = "promql_op.yaml" if context.org_type == "cloud": template = "promql_cloud.yaml" # a list of panel dicts all_metrics = await get_all_metrics(context, template, start, end) checked_metrics_cnt = len(all_metrics) print("perpare_data done, start processing ...") avg_stats, trend_stats = cal_mean_and_trend(all_metrics) if (avg_stats["qps:total"] < 10 and avg_stats["avg_cluster_duration:total"] < 0.01) or ( avg_stats["qps:total"] < 1 ): print("process_data done") advice = { "id": 0, "name": "Nothing to diagnose", "qualified": True, "level": "info", "category": ["perf"], "component": None, "type": "End-to-End", "content": {"key": "metrics_diag_no_workload", "variables": None}, } # 集群无负载，无需诊断。 return { "abnormal": False, "advice": [advice], "stats": { "checkedRules": 1, "scannedMetrics": checked_metrics_cnt, }, } """ if len(avg_stats['tikv_cpu_usage:by_instance']) < 3: print('process_data done') advice = { 'id': 1, 'name': 'Non prod environment', 'qualified': True, 'level': 'info', 'category': [ 'perf' ], 'component': None, 'type': 'End-to-End', 'content': { 'key': 'metrics_diag_no_prod', 'variables': None, }, } # TiKV 实例数小于 3，非生产集群暂无建议。 return {'abnormal': False, 'advice': [advice], 'stats': {"checkedRules": 1, "scannedMetrics": checked_metrics_cnt}} """ act_conn_trend = trend_stats["act_conn:total"] total_conn_trend = trend_stats["connection:total"] qps_trend = trend_stats["qps:total"] tidb_duration_trend = trend_stats["avg_cluster_duration:total"] qps_curve = search(all_metrics, "qps:total").signals.iloc[:, 0].values.tolist() act_conn_curve = search(all_metrics, "act_conn:total").signals.iloc[:, 0].values.tolist() if np.nan in act_conn_curve or np.nan in qps_curve: corr_qps_actconn = np.nan elif max(qps_curve) >= 1e-5: corr_qps_actconn = pearson_correlation_coefficient(act_conn_curve, qps_curve) else: corr_qps_actconn = 1 qps_by_instance = [item for item in avg_stats["qps:by_instance"].values()] act_conn_by_instance = [item for item in avg_stats["act_conn:by_instance"].values()] tidb_duration_by_instance = [ item for item in avg_stats["avg_cluster_duration:by_instance"].values() ] tidb_cpu_by_instance = [item for item in avg_stats["tidb_cpu_usage:by_instance"].values()] tidb_mem_by_instance = [item for item in avg_stats["tidb_memory:by_instance"].values()] tikv_cpu_by_instance = [item for item in avg_stats["tikv_cpu_usage:by_instance"].values()] tikv_mem_by_instance = [item for item in avg_stats["tikv_memory:by_instance"].values()] tikv_disk_by_instance = [item for item in avg_stats["tikv_disk_usage:by_instance"].values()] tikv_disk_iops_by_instance = [item for item in avg_stats["tikv_disk_iops:by_instance"].values()] safepoint_curve = search(all_metrics, "tikv_safepoint:total").signals.iloc[:, 0].values.tolist() safepoint_timestamp = all_metrics[0].signals.index.tolist() safepoint_changetime = [safepoint_timestamp[0]] for i in range(len(safepoint_curve) - 1): if safepoint_curve[i + 1] - safepoint_curve[i] > 0: safepoint_changetime.append(safepoint_timestamp[i + 1]) if len(safepoint_changetime) > 1: diff_timestamp = diff(safepoint_changetime) else: gc_lifetime = 600 # default config diff_timestamp = [0, gc_lifetime, (end - start)] scheduler_panel = search(all_metrics, "pd_schedulers") has_scatter_flag = False balance_schedulers_lack_flag = False for scheduler in scheduler_panel.signals.columns: if "scatter-range-" in scheduler: has_scatter_flag = True if all( x in scheduler_panel.signals.columns for x in [ "balance-hot-region-scheduler", "balance-leader-scheduler", "balance-region-scheduler", ] ): balance_schedulers_lack_flag = False else: balance_schedulers_lack_flag = True raft_engine_curve = ( search(all_metrics, "raft-engine:total").signals.iloc[:, 0].fillna(0).values.tolist() ) raft_message_abnormal = False if "heartbeat" in max( avg_stats["raft_messages:by_type"], key=avg_stats["raft_messages:by_type"].get, ): raft_message_abnormal = True print("process_data done, start diagnosis and output ...") # rules and advice advice = [] checked_rules_cnt = 0 p = Path(__file__).with_name("rules.yaml") with p.open("r") as f: rules = load(f, Loader=Loader) for rule in rules: if rule["switch"]: try: if eval(rule["condition"]): if rule["end2end"]: label = "End-to-End" else: label = "Non-End-to-End" variables = {} if rule["advice"]["variables"] is not None: for variable in rule["advice"]["variables"]: ((k, v),) = variable.items() try: v = eval(v) except Exception as e: print(e) if isinstance(v, float): variables[k] = f"{v:.2f}" else: variables[k] = f"{v}" advice.append( { "id": rule["id"], "name": rule["name"], "qualified": rule["qualified"], "level": rule["level"], "category": rule["category"], "component": rule["component"], "type": label, "content": { "key": rule["advice"]["key"], "variables": variables, }, } ) else: advice.append( { "id": rule["id"], "name": rule["name"], "qualified": rule["qualified"], "level": rule["level"], "category": rule["category"], "component": rule["component"], "type": label, "content": { "key": rule["advice"]["key"], "variables": None, }, } ) checked_rules_cnt += 1 else: checked_rules_cnt += 1 except Exception as e: print("rule exception: ", e) continue start_time = datetime.datetime.fromtimestamp(start + 8 * 3600).strftime("%Y-%m-%d %H:%M:%S") end_time = datetime.datetime.fromtimestamp(end + 8 * 3600).strftime("%Y-%m-%d %H:%M:%S") print( f"Between {start_time} and {end_time}, the following recommendations are available for the cluster {context.cluster_id}." ) for idx, item in enumerate(advice, start=1): print(f" {idx}. [{item['name']}][{item['level']}][{item['type']}] {item['content']}") for metric in all_metrics: if metric.signals.isnull().values.any(): checked_metrics_cnt -= 1 stats = { "checkedRules": checked_rules_cnt, "scannedMetrics": checked_metrics_cnt, } return {"abnormal": len(advice) > 0, "advice": advice, "stats": stats}