# from pandas._libs.lib import dicts_to_array # from compute_framework.decorators import entry import datetime import json import logging from pathlib import Path from typing import Any from context import ClinicContext from utils import ( cal_imbalance, cal_metrics_anomaly, get_panel_data, measurement, set_historical_baseline, threshold, write_annotations, ) # python cli.py call metrics 6860009869741753344 2021-12-21T19:30:00+08:00 2021-12-21T22:30:00+08:00 # http://42.192.1.54:3000/d/JWQcYVY7k/latency-analysis-plan-b?orgId=1&from=1640086021764&to=1640097256952 # python cli.py call metrics 6865911482951002112 2021-12-29T15:16:09+08:00 2021-12-29T17:16:09+08:00 # http://42.192.1.54:3000/d/JWQcYVY7k/latency-analysis-plan-b?orgId=1&from=1640762074885&to=1640769495680 # python cli.py call metrics 6862533764151471104 2022-01-05T05:35:22+08:00 2022-01-05T07:35:22+08:00 # http://42.192.1.54:3000/d/JWQcYVY7k/latency-analysis-plan-b?orgId=1&from=1641332001630&to=1641339409989 # @entry(context='clinic') def handler(context: ClinicContext): logging.basicConfig( level=logging.DEBUG, format="%(asctime)s %(levelname)s %(message)s", filename="./tmp/myapp.log", filemode="w", ) logging.info("Job {} started...".format(str(context.cluster_id))) # Opening JSON file p = Path(__file__).with_name("report_payload.json") with p.open("r") as json_file: report = json.load(json_file) # start_ts = context.start # end_ts = context.stop # start :int = int(datetime.datetime.fromisoformat(start_ts).timestamp()) # end :int = int(datetime.datetime.fromisoformat(end_ts).timestamp()) start: int = int(context.start) end: int = int(context.stop) print("job start, perpare_data ...") # panel_data = get_panel_data(context, 'performance_map_metrics_fast_tune.yaml', start, end) # a list of panel dicts panel_data = get_panel_data( context, "latency_analysis.yaml", start, end ) # a list of panel dicts """ # panel_data looks like: for item in panel_data: print("panel_id: ", item['panel_id']) print("method: ", item['method']) print("object: ", item['object'].name, item['object'].signals.timestamp[-1], item['object'].signals.value[-1]) for candidate in item['candidates']: for signal in candidate.signals: print("candidate: ",candidate.name, " signal: ", signal.name, signal.timestamp[-1], signal.value[-1]) """ print("perpare_data done, start processing ...") annotations = [] bucket_size = 40 # 15 seconds * 40 = 10 minutes sample_time_step = 15 # seconds samples = (end - start) // sample_time_step # tidb_p99_rt:total must be the first one to determine the time ranges tidb_anomaly_seg = [] # tidb_p99_rt:total report_status: dict[str, dict] = {} report_data: list[dict[str, dict]] = [] report_instance: list[dict[str, dict]] = [] for k, v in threshold.items(): report_status[k] = { "value": round(v, 2), "type": "float", "aggr": "first", "timestamp": start, } for panel in panel_data: if panel["object"].name == "tidb_p99_rt:total": object_data = panel["object"] if len(object_data.signals) == 0: print("{} has no signal".format(object_data.name)) continue # set_historical_baseline(context, object_data, 1, end) # set baseline cal_metrics_anomaly(object_data, 20, 0.8) obj = object_data.signals[0] # only one signal # logging.debug("{} outlier: {}".format(obj.name, obj.stationary_anomaly_seg)) # if obj.stationary_anomaly_seg is not []: if len(obj.stationary_anomaly_seg) > 0: # print(f"len obj.startionary_anomaly_seg: {len(obj.stationary_anomaly_seg)}") report["total_latency_good"] = 0 report_status["total_latency_good"] = { "value": 0, "type": "int", "aggr": "first", "timestamp": obj.timestamp[obj.stationary_anomaly_seg[0][0]], } for item in obj.stationary_anomaly_seg: tidb_anomaly_seg.append([obj.timestamp[item[0]], obj.timestamp[item[1] - 1]]) annotations.append( { "panel_id": panel["panel_id"], "obj": panel["object"].name, "annotation": { "start": obj.timestamp[item[0]], "end": obj.timestamp[item[1] - 1], "text": obj.name, "tags": ["anomaly"], }, } ) else: report_status["total_latency_good"] = { "value": 1, "type": "int", "aggr": "first", "timestamp": start, } for seg in tidb_anomaly_seg: start_ts_seg = seg[0] end_ts_seg = seg[1] data = {} data["time"] = { "value": end_ts_seg - start_ts_seg, "type": "duration_seconds", "timestamp": start_ts_seg, } data["get_token_latency_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["parse_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["compile_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["wait_start_tso_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["kv_backoff_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["total_grpc_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["kv_request_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["execute_latency_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["copr_wait_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["copr_handle_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["latch_wait_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["async_write_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["scheduler_command_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["async_snapshot_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } report_data.append(data) ins = {} ins["write_latency_instance"] = { "timestamp": start_ts_seg, "address": [], } ins["copr_latency_instance"] = { "timestamp": start_ts_seg, "address": [], } ins["smallread_latency_instance"] = { "timestamp": start_ts_seg, "address": [], } report_instance.append(ins) # not tidb_p99_rt:total for panel in panel_data: if panel["object"].name != "tidb_p99_rt:total": if panel["method"] == "anomaly" and panel["candidates"] == []: object_data = panel["object"] if len(object_data.signals) == 0: print("{} has no signal".format(object_data.name)) continue # set_historical_baseline(context, object_data, 1, end) # set historical baseline cal_metrics_anomaly(object_data, 20, 0.8) obj = object_data.signals[0] # only one signal # logging.debug("{} outlier: {}".format(obj.name, obj.stationary_anomaly_seg)) # for item in obj.stationary_anomaly_seg: # annotations.append({"panel_id": panel['panel_id'], "obj": panel['object'].name, "annotation": {"start": obj.timestamp[item[0]],"end": obj.timestamp[item[1]-1], "text": obj.name, "tags": ["anomaly"]}}) for seg in tidb_anomaly_seg: for seg_2 in obj.stationary_anomaly_seg: if ( (seg[0] <= obj.timestamp[seg_2[1]]) and (seg[1] >= obj.timestamp[seg_2[0]]) or (obj.timestamp[seg_2[0]] <= seg[1]) and (obj.timestamp[seg_2[1]] >= seg[0]) ): # has overlap annotations.append( { "panel_id": panel["panel_id"], "obj": panel["object"].name, "annotation": { "start": obj.timestamp[seg_2[0]], "end": obj.timestamp[seg_2[1] - 1], "text": obj.name, "tags": ["anomaly"], }, } ) if panel["object"].name == "get_token_duration": data["get_token_latency_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["get_token_latency_good"][ "value" ] = 0 elif panel["object"].name == "parse_duration:general": data["parse_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["parse_good"]["value"] = 0 elif panel["object"].name == "compile_duration:general": data["compile_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["compile_good"][ "value" ] = 0 elif panel["object"].name == "wait_start_tso_duration": data["wait_start_tso_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["wait_start_tso_good"][ "value" ] = 0 elif panel["object"].name == "kv_backoff_duration": data["kv_backoff_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["kv_backoff_good"][ "value" ] = 0 elif panel["object"].name == "tikv_p99_rt:total": data["total_grpc_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["total_grpc_good"][ "value" ] = 0 elif panel["object"].name == "kv_request_duration": data["kv_request_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["kv_request_good"][ "value" ] = 0 elif panel["object"].name == "execute_duration:general": report_data[tidb_anomaly_seg.index(seg)]["execute_latency_good"][ "value" ] = 0 elif panel["object"].name == "copr_wait_duration": data["copr_wait_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["copr_wait_good"][ "value" ] = 0 elif panel["object"].name == "copr_handle_duration": data["copr_handle_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["copr_handle_good"][ "value" ] = 0 elif panel["object"].name == "latch_wait_duration": data["latch_wait_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["latch_wait_good"][ "value" ] = 0 elif panel["object"].name == "async_write_duration": data["async_write_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["async_write_good"][ "value" ] = 0 elif panel["object"].name == "scheduler_command_duration": data["scheduler_command_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["scheduler_command_good"][ "value" ] = 0 elif panel["object"].name == "async_snapshot_duration": data["async_snapshot_good"]["value"] = 0 report_data[tidb_anomaly_seg.index(seg)]["async_snapshot_good"][ "value" ] = 0 elif ( panel["method"] == "anomaly" and panel["candidates"][0].name.find("by_instance") > 0 ): if len(panel["object"].signals) == 0: print("{} has no signal".format(panel["object"].name)) continue obj_data = panel["object"] cal_metrics_anomaly(obj_data, 20, 0.8) obj = obj_data.signals[0] # only one signal for item in obj.stationary_anomaly_seg: annotations.append( { "panel_id": panel["panel_id"], "obj": panel["object"].name, "annotation": { "start": obj.timestamp[item[0]], "end": obj.timestamp[item[1] - 1], "text": obj.name, "tags": ["anomaly"], }, } ) object_data = panel["candidates"][0] # set_historical_baseline(context, object_data, 1, end) # set historical baseline cal_metrics_anomaly(object_data, 20, 0.8) obj = object_data.signals if len(obj) == 0: print("{} has no signal".format(obj.name)) continue for signal in obj: # logging.debug("{} outlier: {}".format(signal.name, signal.stationary_anomaly_seg)) # for sig_seg in signal.stationary_anomaly_seg: # annotations.append({"panel_id": panel['panel_id'], "obj": panel['object'].name, "annotation": {"start": signal.timestamp[sig_seg[0]],"end": signal.timestamp[sig_seg[1]-1], "text": signal.name, "tags": ["anomaly"]}}) for idx, seg in enumerate(tidb_anomaly_seg): # instances = 0 # start_ts = seg[0] # end_ts = seg[1] addr = signal.name for seg_2 in signal.stationary_anomaly_seg: if ( (seg[0] <= signal.timestamp[seg_2[1]]) and (seg[1] >= signal.timestamp[seg_2[0]]) or (signal.timestamp[seg_2[0]] <= seg[1]) and (signal.timestamp[seg_2[1]] >= seg[0]) ): # has overlap # annotations.append({"panel_id": panel['panel_id'], "obj": panel['object'].name, "annotation": {"start": signal.timestamp[seg_2[0]],"end": signal.timestamp[seg_2[1]-1], "text": signal.name, "tags": ["anomaly"]}}) # instances += 1 if "write" in panel["object"].name: # if instances <= 10: # ins['write_latency_instance']['address'].append(addr) report_instance[idx]["write_latency_instance"][ "address" ].append(addr) elif "copr" in panel["object"].name: # if instances <= 10: # ins['copr_latency_instance']['address'].append(addr) report_instance[idx]["copr_latency_instance"]["address"].append( addr ) elif "small" in panel["object"].name: # ins['smallread_latency_instance']['address'].append(addr) report_instance[idx]["smallread_latency_instance"][ "address" ].append(addr) if "write" in panel["object"].name: report_data[idx]["instance_write_latency"] = { "value": len( report_instance[idx]["write_latency_instance"]["address"] ), "type": "int", "timestamp": seg[0], } elif "copr" in panel["object"].name: report_data[idx]["instance_copr_latency"] = { "value": len( report_instance[idx]["copr_latency_instance"]["address"] ), "type": "int", "timestamp": seg[0], } elif "small" in panel["object"].name: report_data[idx]["instance_smallread_latency"] = { "value": len( report_instance[idx]["smallread_latency_instance"]["address"] ), "type": "int", "timestamp": seg[0], } elif panel["method"] == "compare": # if len(panel["object"].signals) == 0: print("{} has no signal".format(panel["object"].name)) continue if len(panel["candidates"][0].signals) == 0: print("{} has no signal".format(panel["candidates"][0].name)) continue if len(panel["candidates"][1].signals) == 0: print("{} has no signal".format(panel["candidates"][1].name)) continue if len(tidb_anomaly_seg) > 0: for idx, item in enumerate(tidb_anomaly_seg): anomaly_start = ( int(item[0] - bucket_size * sample_time_step * 1) if item[0] - bucket_size * sample_time_step * 1 >= start else start ) anomaly_end = int(item[1] - sample_time_step) anomaly_start_pos = ( panel["candidates"][0].signals[0].timestamp.index(anomaly_start) ) anomaly_end_pos = ( panel["candidates"][0].signals[0].timestamp.index(anomaly_end) ) total_grpc = sum( panel["object"].signals[0].value[anomaly_start_pos:anomaly_end_pos] ) write_rpc = sum( panel["candidates"][0] .signals[0] .value[anomaly_start_pos:anomaly_end_pos] ) copr_rpc = sum( panel["candidates"][1] .signals[0] .value[anomaly_start_pos:anomaly_end_pos] ) write_ratio = write_rpc / total_grpc copr_ratio = copr_rpc / total_grpc smallread_ratio = 1 - write_ratio - copr_ratio report_data[idx]["grpc_proportion_write"] = { "value": round((write_ratio * 100), 2), "type": "float", "timestamp": item[0], } report_data[idx]["grpc_proportion_copr"] = { "value": round((copr_ratio * 100), 2), "type": "float", "timestamp": item[0], } report_data[idx]["grpc_proportion_small_read"] = { "value": round((smallread_ratio * 100), 2), "type": "float", "timestamp": item[0], } elif panel["method"] == "imbalance": if len(panel["object"].signals) == 0: print("{} has no signal".format(panel["object"].name)) continue if len(tidb_anomaly_seg) > 0: object_data = panel["object"] # set_historical_baseline(context, object_data, 1, end) # set historical baseline if "baseline" in object_data.signals[0].details.keys(): baseline = [] for sig in object_data.signals: if "baseline" in sig.details.keys(): baseline.append(sig.details["baseline"]) _, baseline_stats = cal_imbalance(baseline, 0.8, 9) else: baseline_stats = { "p10": threshold["imbalance"], "p20": threshold["imbalance"], "p50": threshold["imbalance"], "p95": threshold["imbalance"], "avg": threshold["imbalance"], "std": 0, } for idx, item in enumerate(tidb_anomaly_seg): start_ts = item[0] anomaly_start = ( int(item[0] - bucket_size * sample_time_step * 1) if item[0] - bucket_size * sample_time_step * 1 >= start else start ) anomaly_end = int(item[1] - sample_time_step) anomaly_start_pos = ( panel["object"].signals[0].timestamp.index(anomaly_start) ) anomaly_end_pos = panel["object"].signals[0].timestamp.index(anomaly_end) cakes = [] for signal in panel["object"].signals: cakes.append(signal.get_slice(anomaly_start_pos, anomaly_end_pos)) imbalance, stats = cal_imbalance(cakes, 0.8, 9) # 0.4 maybe too small # for it in imbalance: # annotations.append({"panel_id": panel['panel_id'], "obj": panel['object'].name, "candidate": "None", "annotation": {"start": it[0],"end": it[1], "text": signal.name, "tags": ["imbalance"]}}) # print("imbalance P20: ", stats['p20'], "imbalance P50: ", stats['p50'], " AVG: ", stats['avg']) logging.debug("baseline_stats['p20']: {}".format(baseline_stats["p20"])) if panel["object"].name == "write-rpc:by_instance": report_data[idx]["imbalance_write_ratio"] = { "value": round(stats["p20"], 2), "type": "float", "timestamp": start_ts, } report_data[idx]["imbalance_write_baseline_1d"] = { "value": baseline_stats["p20"], "type": "float", "timestamp": start_ts, } report_data[idx]["imbalance_write_baseline_1h"] = { "value": baseline_stats["p20"], "type": "float", "timestamp": start_ts, } trend = 0 if stats["p20"] < baseline_stats["p20"] * 0.8: trend = 2 elif stats["p20"] > baseline_stats["p20"] * 1.2: trend = 1 else: trend = 0 report_data[idx]["imbalance_write_trend"] = { "value": trend, "type": "int", "timestamp": start_ts, } elif panel["object"].name == "copr-rpc:by_instance": report_data[idx]["imbalance_copr_ratio"] = { "value": round(stats["p20"], 2), "type": "float", "timestamp": start_ts, } report_data[idx]["imbalance_copr_baseline_1d"] = { "value": baseline_stats["p20"], "type": "float", "timestamp": start_ts, } report_data[idx]["imbalance_copr_baseline_1h"] = { "value": baseline_stats["p20"], "type": "float", "timestamp": start_ts, } trend = 0 if stats["p20"] < baseline_stats["p20"] * 0.8: trend = 2 elif stats["p20"] > baseline_stats["p20"] * 1.2: trend = 1 else: trend = 0 report_data[idx]["imbalance_copr_trend"] = { "value": trend, "type": "int", "timestamp": start_ts, } elif panel["object"].name == "small-read-rpc:by_instance": report_data[idx]["imbalance_small_read_ratio"] = { "value": round(stats["p20"], 2), "type": "float", "timestamp": start_ts, } report_data[idx]["imbalance_small_read_baseline_1d"] = { "value": baseline_stats["p20"], "type": "float", "timestamp": start_ts, } report_data[idx]["imbalance_small_read_baseline_1h"] = { "value": baseline_stats["p20"], "type": "float", "timestamp": start_ts, } trend = 0 if stats["p20"] < baseline_stats["p20"] * 0.8: trend = 2 elif stats["p20"] > baseline_stats["p20"] * 1.2: trend = 1 else: trend = 0 report_data[idx]["imbalance_small_read_trend"] = { "value": trend, "type": "int", "timestamp": start_ts, } for item in annotations: logging.debug("Annotation: {}".format(item)) logging.debug("{} annotations are written.".format(len(annotations))) print("processing done") print("writing report ...") # with open("./tmp/payload.json", "w") as outfile: # json.dump(report, outfile, indent=4) # logging.debug("report: {}".format(report)) with open("./tmp/report_data.json", "w") as outfile: json.dump(report_data, outfile, indent=4) # logging.debug("report: {}".format(report)) with open("./tmp/report_instance.json", "w") as outfile: json.dump(report_instance, outfile, indent=4) # logging.debug("report: {}".format(report)) print("writing annotation data ...") write_annotations(context, annotations) print("writing report data ...") # write_report(context, report) for key, value in report_status.items(): payload = { "timestamp": value["timestamp"], "measurement": f"{measurement}_dt", "tidb_cluster": context.cluster_id, "fields": {key: value["value"]}, "tags": {"type": value["type"], "aggr": value["aggr"]}, } resp = context.write_timeseries_data_v2(payload=payload) if not resp.ok: print(f"payload: {payload}") print( f"write_timeseries_data failed, status_code: {resp.status_code} reason: {resp.reason}" ) for i, item in enumerate(report_data): if i > 2: continue for key, value in item.items(): payload = { "timestamp": value["timestamp"], "measurement": f"{measurement}_dt", "tidb_cluster": context.cluster_id, "fields": {key: value["value"]}, "tags": {"type": value["type"]}, } resp = context.write_timeseries_data_v2(payload=payload) if not resp.ok: print(f"payload: {payload}") print( f"write_timeseries_data failed, status_code: {resp.status_code} reason: {resp.reason}" ) for i, item in enumerate(report_instance): if i > 2: continue for key, value in item.items(): timestamp = int(value["timestamp"]) address: list[str] = value["address"] for addr in address: payload = { "timestamp": timestamp, "measurement": f"{measurement}_dt", "tidb_cluster": context.cluster_id, "fields": {key: 1}, "tags": {"type": "address", "instance": addr}, } resp = context.write_timeseries_data_v2(payload=payload) if not resp.ok: print(f"payload: {payload}") print( f"write_timeseries_data failed, status_code: {resp.status_code} reason: {resp.reason}" ) # print("flushing data ...") # flush_data(context) logging.info("Done") return {"status": "done"}