# from typing import Any import datetime from pathlib import Path from context import ClinicContext from job_latency import handler as p99_handler from utils_avg import ( avg_threshold, cal_imbalance, cal_metrics_anomaly, get_panel_data, measurement, set_historical_baseline, 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/nPzC1XYnz/average-latency-analysis?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/nPzC1XYnz/average-latency-analysis?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/nPzC1XYnz/average-latency-analysis?orgId=1&from=1641332001630&to=1641339409989 def search_panel(name, dicts): return next((item for item in dicts if item["panel_id"] == name), False) def handler(context: ClinicContext): # 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 ...") template = "latency_avg_analysis.yaml" if context.org_type == "cloud": template = "latency_avg_analysis_cloud.yaml" panel_data = get_panel_data(context, template, 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 panel in panel_data: # print(panel['object'].name, " ", len(panel['object'].signals)) # step 1: conn_idle_duration vs tidb_latency # step 1.1: tidb_latency tidb_latency_panel = search_panel(2, panel_data) if len(tidb_latency_panel["object"].signals) == 0: return {"status": "TiDB latency has no value."} else: cal_metrics_anomaly( tidb_latency_panel["object"], 21, 0.3 ) # 4,0.8的话，11 个单独的峰；9,0.6的话，就还是只有 1 个，20，0.8 的话没有了 tidb_latency = tidb_latency_panel["object"].signals[0] if len(tidb_latency.stationary_anomaly_seg) > 0: report_status["total_latency_good"] = { "value": 0, "type": "int", "aggr": "first", "timestamp": tidb_latency.timestamp[tidb_latency.stationary_anomaly_seg[0][0]], } for item in tidb_latency.stationary_anomaly_seg: tidb_anomaly_seg.append( [ tidb_latency.timestamp[item[0]], tidb_latency.timestamp[item[1] - 1], ] ) annotations.append( { "panel_id": tidb_latency_panel["panel_id"], "obj": tidb_latency_panel["object"].name, "annotation": { "start": tidb_latency.timestamp[item[0]], "end": tidb_latency.timestamp[item[1] - 1], "text": tidb_latency.name, "tags": ["anomaly"], }, } ) else: report_status["total_latency_good"] = { "value": 1, "type": "int", "aggr": "first", "timestamp": start, } panel_data.remove(tidb_latency_panel) for seg in tidb_anomaly_seg: start_ts_seg = seg[0] end_ts_seg = seg[1] data = {} # conn_idle_good -> report_status data["conn_prop"] = { "value": 0.0, "type": "float", "timestamp": start_ts_seg, } data["qps_prop"] = { "value": 0.0, "type": "float", "timestamp": start_ts_seg, } data["cps_prop_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } # cps_prop_good_time_1,2,3 high=good <0.2 not good data["plan_cache_prop"] = { "value": 2, "type": "int", "timestamp": start_ts_seg, } # plan_cache_prop_time_1,2,3 0: all zero, 1: low, 2: high data["store_latency_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } data["apply_latency_good"] = { "value": 1, "type": "int", "timestamp": start_ts_seg, } 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) for k, v in avg_threshold.items(): report_status[k] = { "value": round(v, 2), "type": "float", "aggr": "first", "timestamp": start, } # step 1.2: conn_idle conn_idle_duration_panel = search_panel(98, panel_data) if len(conn_idle_duration_panel["object"].signals) == 0: conn_idle_duration_not_in_txn = conn_idle_duration_panel["candidates"][0] if len(conn_idle_duration_not_in_txn.signals) == 0: conn_idle = None else: cal_metrics_anomaly(conn_idle_duration_not_in_txn, 21, 0.3) conn_idle = conn_idle_duration_not_in_txn.signals[0] else: conn_idle_duration_in_txn = conn_idle_duration_panel["object"] cal_metrics_anomaly(conn_idle_duration_in_txn, 21, 0.3) conn_idle = conn_idle_duration_in_txn.signals[0] if conn_idle is not None: if len(conn_idle.stationary_anomaly_seg) > 0: for item in conn_idle.stationary_anomaly_seg: annotations.append( { "panel_id": conn_idle_duration_panel["panel_id"], "obj": conn_idle_duration_panel["object"].name, "annotation": { "start": conn_idle.timestamp[item[0]], "end": conn_idle.timestamp[item[1] - 1], "text": conn_idle.name, "tags": ["anomaly"], }, } ) report_status["conn_idle_good"] = { "value": 0, "type": "int", "aggr": "first", "timestamp": start, } # to be checked else: report_status["conn_idle_good"] = { "value": 1, "type": "int", "aggr": "first", "timestamp": start, } # to be checked panel_data.remove(conn_idle_duration_panel) # step 1.3: other evidences # step 1.3.1: act / total connection conn_proportion_panel = search_panel(100, panel_data) if ( len(conn_proportion_panel["object"].signals) > 0 and len(conn_proportion_panel["candidates"][0].signals) > 0 ): # conn_prop = conn_proportion_panel['candidates'][0].signals[0] / conn_proportion_panel['object'].signals[0] # print("conn prop: ", max(conn_prop.value)) for idx, item in enumerate(tidb_anomaly_seg): anomaly_start = int(item[0]) anomaly_end = int(item[1] - sample_time_step) anomaly_start_pos = ( conn_proportion_panel["candidates"][0].signals[0].timestamp.index(anomaly_start) ) anomaly_end_pos = ( conn_proportion_panel["candidates"][0].signals[0].timestamp.index(anomaly_end) ) conn_prop = sum( conn_proportion_panel["candidates"][0] .signals[0] .value[anomaly_start_pos:anomaly_end_pos] ) / sum( conn_proportion_panel["object"].signals[0].value[anomaly_start_pos:anomaly_end_pos] ) report_data[idx]["conn_prop"] = { "value": round((conn_prop * 100), 2), "type": "float", "timestamp": item[0], } panel_data.remove(conn_proportion_panel) # step 1.3.2: failed / total qps qps_panel = search_panel(102, panel_data) if len(qps_panel["object"].signals) > 0 and len(qps_panel["candidates"][0].signals) > 0: # qps_prop = qps_panel['candidates'][0].signals[0] / qps_panel['object'].signals[0] # qps_panel['object'].signals = [qps_prop] # not good # cal_metrics_anomaly(qps_panel['object'], 21, 0.3) for idx, item in enumerate(tidb_anomaly_seg): anomaly_start = int(item[0]) anomaly_end = int(item[1] - sample_time_step) anomaly_start_pos = qps_panel["candidates"][0].signals[0].timestamp.index(anomaly_start) anomaly_end_pos = qps_panel["candidates"][0].signals[0].timestamp.index(anomaly_end) qps_prop = sum( qps_panel["candidates"][0].signals[0].value[anomaly_start_pos:anomaly_end_pos] ) / sum(qps_panel["object"].signals[0].value[anomaly_start_pos:anomaly_end_pos]) report_data[idx]["qps_prop"] = { "value": round((qps_prop * 100), 2), "type": "float", "timestamp": item[0], } panel_data.remove(qps_panel) # step 1.4: check if latency mainly in tidb cluster """ if len(tidb_anomaly_seg) == 0: if conn_idle is not None: if len(conn_idle.stationary_anomaly_seg) > 0: return {"status":"Application has latency."} else: return {"status":"No performance issue found."} else: return {"status":"Not sure if application has performance issue."} # 如果 tidb latency 有问题，暂时没管 conn_idle 和它的时间重叠区间，也不再单独声明应用的问题，应用有问题统一说一下如图示区域应用可能存在问题。 """ # without 4 panel in step 1 for panel in panel_data: 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, 21, 0.3) # if object_data.name == 'latch_wait_duration': # print("latch_wait_duration: {}".format(object_data.signals[0].value[obj_data.signals[0].stationary_anomaly_seg[0][0]:obj_data.signals[0].stationary_anomaly_seg[0][1]])) obj = object_data.signals[0] # only one signal 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": report_data[tidb_anomaly_seg.index(seg)]["get_token_latency_good"][ "value" ] = 0 elif panel["object"].name == "parse_duration:general": report_data[tidb_anomaly_seg.index(seg)]["parse_good"]["value"] = 0 elif panel["object"].name == "compile_duration:general": report_data[tidb_anomaly_seg.index(seg)]["compile_good"]["value"] = 0 elif panel["object"].name == "wait_start_tso_duration": report_data[tidb_anomaly_seg.index(seg)]["wait_start_tso_good"][ "value" ] = 0 elif panel["object"].name == "kv_backoff_duration": report_data[tidb_anomaly_seg.index(seg)]["kv_backoff_good"]["value"] = 0 elif panel["object"].name == "tikv_p99_rt:total": report_data[tidb_anomaly_seg.index(seg)]["total_grpc_good"]["value"] = 0 elif panel["object"].name == "kv_request_duration": 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": report_data[tidb_anomaly_seg.index(seg)]["copr_wait_good"]["value"] = 0 elif panel["object"].name == "copr_handle_duration": report_data[tidb_anomaly_seg.index(seg)]["copr_handle_good"][ "value" ] = 0 elif panel["object"].name == "latch_wait_duration": report_data[tidb_anomaly_seg.index(seg)]["latch_wait_good"]["value"] = 0 elif panel["object"].name == "async_write_duration": report_data[tidb_anomaly_seg.index(seg)]["async_write_good"][ "value" ] = 0 elif panel["object"].name == "scheduler_command_duration": report_data[tidb_anomaly_seg.index(seg)]["scheduler_command_good"][ "value" ] = 0 elif panel["object"].name == "async_snapshot_duration": report_data[tidb_anomaly_seg.index(seg)]["async_snapshot_good"][ "value" ] = 0 elif panel["object"].name == "apply": report_data[tidb_anomaly_seg.index(seg)]["apply_latency_good"][ "value" ] = 0 elif panel["object"].name == "store": report_data[tidb_anomaly_seg.index(seg)]["store_latency_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, 21, 0.3) # print("instance: ",obj_data.signals[0].name," anomaly: ", obj_data.signals[0].stationary_anomaly_seg) 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, 21, 0.3) # print("instance: ",object_data.signals[0].name," anomaly: ", object_data.signals[0].stationary_anomaly_seg) obj = object_data.signals if len(obj) == 0: # print("{} has no signal".format(obj.name)) continue for signal in obj: 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 # print(panel['object'].name, signal.name) 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 panel["object"].name == "grpc:total": 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["candidates"][0].name == "stmt_execute": if len(panel["candidates"][0].signals) > 0: stmt_execute = panel["candidates"][0].signals[0] if panel["object"].name == "plan_cache": if len(panel["object"].signals) > 0: plan_cache = panel["object"].signals[0] plan_cache_prop = plan_cache / stmt_execute for idx, item in enumerate(tidb_anomaly_seg): anomaly_start = int(item[0]) 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) ) plan_cache_avg = sum( plan_cache_prop.value[anomaly_start_pos:anomaly_end_pos] ) / (anomaly_end_pos - anomaly_start_pos + 1) if plan_cache_avg == 0: report_data[idx]["plan_cache_prop"] = { "value": 0, "type": "int", "timestamp": item[0], } elif plan_cache_avg < 0.1 and plan_cache_avg > 0: report_data[idx]["plan_cache_prop"] = { "value": 1, "type": "int", "timestamp": item[0], } elif panel["object"].name == "cps_total": if len(panel["object"].signals) > 0: cps_total = panel["object"].signals[0] cps_prop = stmt_execute / cps_total for idx, item in enumerate(tidb_anomaly_seg): anomaly_start = int(item[0]) 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) ) plan_cache_avg = sum( cps_prop.value[anomaly_start_pos:anomaly_end_pos] ) / (anomaly_end_pos - anomaly_start_pos + 1) if plan_cache_avg < 0.2: report_data[idx]["cps_prop_good"] = { "value": 0, "type": "int", "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": avg_threshold["imbalance"], "p20": avg_threshold["imbalance"], "p50": avg_threshold["imbalance"], "p95": avg_threshold["imbalance"], "avg": 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 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, } print("processing done") 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(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(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(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("running p99 job...") # p99_handler(context) return {"status": "done"}