import asyncio import json import operator def flatten_map(input_map, separator=".", prefix=""): return ( { prefix + separator + k if prefix else k: v for kk, vv in input_map.items() for k, v in flatten_map(vv, separator, kk).items() } if isinstance(input_map, dict) else {prefix: input_map} ) def diff(map1, map2, component_type="TiDB"): # flatten map1 and map2 flatten_map1 = flatten_map(map1) flatten_map2 = flatten_map(map2) # get diff results diff_results = [] for key1, value1 in flatten_map1.items(): if key1 in flatten_map2 and not operator.eq(value1, flatten_map2[key1]): diff_results.append( { "values": { "components": component_type, "configurationItems": key1, "t1Value": value1, "t2Value": flatten_map2[key1], }, "subValues": None, } ) return diff_results def diff_configuration(t1_configuration, t2_configuration): try: if t1_configuration["status"] == "failed": return ( False, "Get t1 range configuration failed: %s!" % t1_configuration["data"]["errorMessage"], ) if t2_configuration["status"] == "failed": return ( False, "Get t2 range configuration failed: %s!" % t2_configuration["data"]["errorMessage"], ) diff_results = [] # diff pd config t1_pd_config = json.loads(t1_configuration["data"]["pd_config"]) t2_pd_config = json.loads(t2_configuration["data"]["pd_config"]) diff_results += diff(t1_pd_config, t2_pd_config, "PD") # diff tidb config t1_tidb_config = json.loads(t1_configuration["data"]["tidb_config"]) t2_tidb_config = json.loads(t2_configuration["data"]["tidb_config"]) diff_results += diff(t1_tidb_config, t2_tidb_config, "TiDB") # diff tikv config t1_tikv_config = json.loads(t1_configuration["data"]["tikv_config"]) t2_tikv_config = json.loads(t2_configuration["data"]["tikv_config"]) diff_results += diff(t1_tikv_config, t2_tikv_config, "TiKV") # diff tiflash config t1_tiflash_config = json.loads(t1_configuration["data"]["tiflash_config"]) t2_tiflash_config = json.loads(t2_configuration["data"]["tiflash_config"]) diff_results += diff(t1_tiflash_config, t2_tiflash_config, "TiFlash") return True, diff_results except Exception as e: print(e) raise e def diff_servers(t1_servers, t2_servers, component_type, port_field="port"): diff_servers_info = [] t1_servers_count = len(t1_servers) t2_servers_count = len(t2_servers) diff_servers_count = t1_servers_count if t1_servers_count < t2_servers_count: diff_servers_count = t2_servers_count for index in range(diff_servers_count): if index >= t1_servers_count: diff_servers_info.append( { "values": { "components": component_type, "t1Instances": "", "t2Instances": t2_servers[index]["host"] + ":" + str(t2_servers[index][port_field]), }, "subValues": None, } ) elif index >= t2_servers_count: diff_servers_info.append( { "values": { "components": component_type, "t1Instances": t1_servers[index]["host"] + ":" + str(t1_servers[index][port_field]), "t2Instances": "", }, "subValues": None, } ) else: diff_servers_info.append( { "values": { "components": component_type, "t1Instances": t1_servers[index]["host"] + ":" + str(t1_servers[index][port_field]), "t2Instances": t2_servers[index]["host"] + ":" + str(t2_servers[index][port_field]), }, "subValues": None, } ) return diff_servers_info def diff_topology_info(t1_configuration, t2_configuration): try: if t1_configuration["status"] == "failed": return ( False, "Get t1 range configuration failed: %s!" % t1_configuration["data"]["errorMessage"], ) if t2_configuration["status"] == "failed": return ( False, "Get t2 range configuration failed: %s!" % t2_configuration["data"]["errorMessage"], ) diff_topology_infos = [] # diff topology info t1_cluster_info = json.loads(t1_configuration["data"]["cluster_info"]) t2_cluster_info = json.loads(t2_configuration["data"]["cluster_info"]) diff_topology_infos += diff_servers( t1_cluster_info["pd_servers"], t2_cluster_info["pd_servers"], "PD", "port", ) diff_topology_infos += diff_servers( t1_cluster_info["tidb_servers"], t2_cluster_info["tidb_servers"], "TiDB", "port", ) diff_topology_infos += diff_servers( t1_cluster_info["tikv_servers"], t2_cluster_info["tikv_servers"], "TiKV", "port", ) diff_topology_infos += diff_servers( t1_cluster_info["tiflash_servers"], t2_cluster_info["tiflash_servers"], "TiFlash", "port", ) return True, diff_topology_infos except Exception as e: print(e) raise e def top_dict(input_dict, n): temp_list = sorted(input_dict.items(), key=lambda item: item[1], reverse=True) if len(temp_list) > n: topn_list = temp_list[:n] else: topn_list = temp_list topn_dict = {} for l in topn_list: topn_dict[l[0]] = l[1] return topn_dict def has_metrics(metrics): stats = metrics["data"]["stats"] has_metrics_data = False for stats_index in range(len(stats)): if stats[stats_index]["label"] == "total": stats_name = stats[stats_index]["metrics"] + ":" + stats[stats_index]["label"] if stats[stats_index]["value"][stats_name] != 0.0: has_metrics_data = True elif stats[stats_index]["label"] == "by_type": if len(stats[stats_index]["value"]) > 1: has_metrics_data = True elif stats[stats_index]["label"] == "by_instance": if len(stats[stats_index]["value"]) > 1: has_metrics_data = True return has_metrics_data def metrics_value(value, unit): if value is None: return None if unit == "1": return "%.2f" % value elif unit == "s": units = ["s", "ms", "us"] size = 1000.0 for i in range(len(units)): if value > 1 or units[i] == "us": return "%.2f%s" % (value, units[i]) value = value * size elif unit == "Bytes": units = ["B", "KB", "MB", "GB", "TB", "PB"] size = 1024.0 for i in range(len(units)): if value / size < 1: return "%.2f%s" % (value, units[i]) value = value / size elif unit == "%": return "%.2f%%" % (value) elif unit == "Bps": units = ["B/s", "KB/s", "MB/s", "GB/s", "TB/s", "PB/s"] size = 1024.0 for i in range(len(units)): if value / size < 1: return "%.2f%s" % (value, units[i]) value = value / size def diff_one_empty_sub_metrics(metrics, stats_name, not_empty_key): diff_metrics_results = [] stats = metrics["data"]["stats"] for stats_index in range(len(stats)): current_stats_name = stats[stats_index]["metrics"] + ":" + stats[stats_index]["label"] if stats_name == current_stats_name: stats_unit = stats[stats_index]["unit"] stats_value = stats[stats_index]["value"][stats_name] if not isinstance(stats_value, dict): return diff_metrics_results if stats[stats_index]["label"] == "by_type": top_stats_value = top_dict(stats_value, 5) else: top_stats_value = stats_value for key, value in top_stats_value.items(): sub_stats_value = stats[stats_index]["value"][stats_name].get(key) diff_metrics_results.append( { "values": { "components": stats[stats_index]["component"], "metrics": stats[stats_index]["metrics"] + "." + key, "label": stats[stats_index]["label"], "t1AverageValue": metrics_value(sub_stats_value, stats_unit) if not_empty_key == "t1" else None, "t2AverageValue": metrics_value(sub_stats_value, stats_unit) if not_empty_key == "t2" else None, "diff": None, "description": stats[stats_index]["description"], "grafanaLocation": stats[stats_index]["grafana"], }, "subValues": None, } ) return diff_metrics_results def diff_one_empty_metrics(metrics, metric_type, not_empty_key): diff_metrics_results = [] stats = metrics["data"]["stats"] for stats_index in range(len(stats)): if metric_type not in stats[stats_index]["type"]: continue stats_name = stats[stats_index]["metrics"] + ":" + stats[stats_index]["label"] stats_unit = stats[stats_index]["unit"] stats_value = stats[stats_index]["value"][stats_name] if isinstance(stats_value, dict) and stats[stats_index]["label"] == "by_type": top_stats_value = top_dict(stats_value, 5) else: top_stats_value = stats_value if isinstance(stats_value, dict) and (not stats[stats_index]["only_sub"]): for key, value in top_stats_value.items(): sub_stats_value = stats[stats_index]["value"][stats_name].get(key) diff_metrics_results.append( { "values": { "components": stats[stats_index]["component"], "metrics": stats[stats_index]["metrics"] + "." + key, "label": stats[stats_index]["label"], "t1AverageValue": metrics_value(sub_stats_value, stats_unit) if not_empty_key == "t1" else None, "t2AverageValue": metrics_value(sub_stats_value, stats_unit) if not_empty_key == "t2" else None, "diff": None, "description": stats[stats_index]["description"], "grafanaLocation": stats[stats_index]["grafana"], }, "subValues": None, } ) elif not isinstance(stats_value, dict): if len(stats[stats_index]["sub_metrics"]) > 0: sub_values = [] for key in stats[stats_index]["sub_metrics"]: sub_values += diff_one_empty_sub_metrics(metrics, key, not_empty_key) diff_metrics_results.append( { "values": { "components": stats[stats_index]["component"], "metrics": stats[stats_index]["metrics"], "label": stats[stats_index]["label"], "t1AverageValue": metrics_value(stats_value, stats_unit) if not_empty_key == "t1" else None, "t2AverageValue": metrics_value(stats_value, stats_unit) if not_empty_key == "t1" else None, "diff": None, "description": stats[stats_index]["description"], "grafanaLocation": stats[stats_index]["grafana"], }, "subValues": sub_values, } ) else: diff_metrics_results.append( { "values": { "components": stats[stats_index]["component"], "metrics": stats[stats_index]["metrics"], "label": stats[stats_index]["label"], "t1AverageValue": metrics_value(stats_value, stats_unit) if not_empty_key == "t1" else None, "t2AverageValue": metrics_value(stats_value, stats_unit) if not_empty_key == "t2" else None, "diff": None, "description": stats[stats_index]["description"], "grafanaLocation": stats[stats_index]["grafana"], }, "subValues": None, } ) return diff_metrics_results def get_perf_metrics_modules(metrics): perf_metrics_modules = set() stats = metrics["data"]["stats"] for stats_index in range(len(stats)): for metric_type in stats[stats_index]["type"]: if metric_type.startswith("Perf"): perf_metrics_modules.add(metric_type) return list(perf_metrics_modules) def diff_perf_metrics(t1_metrics, t2_metrics): if t1_metrics["status"] == "failed": raise Exception("Get t1 range metrics failed: %s!" % t1_metrics["data"]["errorMessage"]) if t2_metrics["status"] == "failed": raise Exception("Get t2 range metrics failed: %s!" % t2_metrics["data"]["errorMessage"]) if not (has_metrics(t1_metrics) or has_metrics(t2_metrics)): return [] if has_metrics(t1_metrics): perf_metrics_modules = get_perf_metrics_modules(t1_metrics) else: perf_metrics_modules = get_perf_metrics_modules(t2_metrics) diff_metrics_results = {} for module in perf_metrics_modules: diff_metrics_results[module] = diff_metrics(t1_metrics, t2_metrics, module) return diff_metrics_results def diff_sub_metrics(t1_metrics, t2_metrics, stats_name): diff_metrics_results = [] t1_stats = t1_metrics["data"]["stats"] t2_stats = t2_metrics["data"]["stats"] for stats_index in range(len(t1_stats)): current_stats_name = t1_stats[stats_index]["metrics"] + ":" + t1_stats[stats_index]["label"] if stats_name == current_stats_name: stats_unit = t1_stats[stats_index]["unit"] stats_value1 = t1_stats[stats_index]["value"][stats_name] stats_value2 = t2_stats[stats_index]["value"][stats_name] if not isinstance(stats_value1, dict): return diff_metrics_results if t1_stats[stats_index]["label"] == "by_type": top_stats_value = top_dict(stats_value2, 5) else: top_stats_value = stats_value2 for key, value in top_stats_value.items(): sub_stats_value1 = t1_stats[stats_index]["value"][stats_name].get(key) sub_stats_value2 = t2_stats[stats_index]["value"][stats_name].get(key) if sub_stats_value1 is not None and sub_stats_value2 is not None: diff_value = (sub_stats_value2 - sub_stats_value1) / ( sub_stats_value1 + 1e-10 ) # avoid zero division if ( metrics_value(sub_stats_value2, stats_unit).startswith("0.00") and sub_stats_value1 < 0.1 ): # if t2 value is shown as 0.00, and t1 value is small, set diff value to 0 diff_value = 0 else: diff_value = None diff_metrics_results.append( { "values": { "components": t1_stats[stats_index]["component"], "metrics": t1_stats[stats_index]["metrics"] + "." + key, "label": t1_stats[stats_index]["label"], "t1AverageValue": metrics_value(sub_stats_value1, stats_unit), "t2AverageValue": metrics_value(sub_stats_value2, stats_unit), "diff": diff_value, "description": t1_stats[stats_index]["description"], "grafanaLocation": t1_stats[stats_index]["grafana"], }, "subValues": None, } ) return diff_metrics_results def diff_metrics(t1_metrics, t2_metrics, metric_type): if t1_metrics["status"] == "failed": raise Exception("Get t1 range metrics failed: %s!" % t1_metrics["data"]["errorMessage"]) if t2_metrics["status"] == "failed": raise Exception("Get t2 range metrics failed: %s!" % t2_metrics["data"]["errorMessage"]) if not (has_metrics(t1_metrics) or has_metrics(t2_metrics)): return [] if has_metrics(t1_metrics) and (not has_metrics(t2_metrics)): return diff_one_empty_metrics(t1_metrics, metric_type, "t1") if (not has_metrics(t1_metrics)) and has_metrics(t2_metrics): return diff_one_empty_metrics(t2_metrics, metric_type, "t2") diff_metrics_results = [] t1_stats = t1_metrics["data"]["stats"] t2_stats = t2_metrics["data"]["stats"] for stats_index in range(len(t1_stats)): if metric_type in t1_stats[stats_index]["type"]: stats_name = t1_stats[stats_index]["metrics"] + ":" + t1_stats[stats_index]["label"] stats_unit = t1_stats[stats_index]["unit"] stats_value1 = t1_stats[stats_index]["value"][stats_name] stats_value2 = t2_stats[stats_index]["value"][stats_name] if isinstance(stats_value1, dict) and t1_stats[stats_index]["label"] == "by_type": top_stats_value = top_dict(stats_value2, 3) else: top_stats_value = stats_value2 if isinstance(stats_value1, dict) and (not t1_stats[stats_index]["only_sub"]): for key, value in top_stats_value.items(): sub_stats_value1 = t1_stats[stats_index]["value"][stats_name].get(key) sub_stats_value2 = t2_stats[stats_index]["value"][stats_name].get(key) if sub_stats_value1 is not None and sub_stats_value2 is not None: diff_value = (sub_stats_value2 - sub_stats_value1) / ( sub_stats_value1 + 1e-10 ) # avoid zero division if ( metrics_value(sub_stats_value2, stats_unit).startswith("0.00") and sub_stats_value1 < 0.1 ): # if t2 value is shown as 0.00, and t1 value is small, set diff value to 0 diff_value = 0 else: diff_value = None diff_metrics_results.append( { "values": { "components": t1_stats[stats_index]["component"], "metrics": t1_stats[stats_index]["metrics"] + "." + key, "label": t1_stats[stats_index]["label"], "t1AverageValue": metrics_value(sub_stats_value1, stats_unit), "t2AverageValue": metrics_value(sub_stats_value2, stats_unit), "diff": diff_value, "description": t1_stats[stats_index]["description"], "grafanaLocation": t1_stats[stats_index]["grafana"], }, "subValues": None, } ) elif not isinstance(stats_value1, dict): if stats_value1 is not None and stats_value2 is not None: diff_value = (stats_value2 - stats_value1) / ( stats_value1 + 1e-10 ) # avoid zero division if ( metrics_value(stats_value2, stats_unit).startswith("0.00") and stats_value1 < 0.1 ): # if t2 value is shown as 0.00, and t1 value is small, set diff value to 0 diff_value = 0 else: diff_value = None if len(t1_stats[stats_index]["sub_metrics"]) > 0: sub_values = [] for key in t1_stats[stats_index]["sub_metrics"]: sub_values += diff_sub_metrics(t1_metrics, t2_metrics, key) diff_metrics_results.append( { "values": { "components": t1_stats[stats_index]["component"], "metrics": t1_stats[stats_index]["metrics"], "label": t1_stats[stats_index]["label"], "t1AverageValue": metrics_value(stats_value1, stats_unit), "t2AverageValue": metrics_value(stats_value2, stats_unit), "diff": diff_value, "description": t1_stats[stats_index]["description"], "grafanaLocation": t1_stats[stats_index]["grafana"], }, "subValues": sub_values, } ) else: diff_metrics_results.append( { "values": { "components": t1_stats[stats_index]["component"], "metrics": t1_stats[stats_index]["metrics"], "label": t1_stats[stats_index]["label"], "t1AverageValue": metrics_value(stats_value1, stats_unit), "t2AverageValue": metrics_value(stats_value2, stats_unit), "diff": diff_value, "description": t1_stats[stats_index]["description"], "grafanaLocation": t1_stats[stats_index]["grafana"], }, "subValues": None, } ) return diff_metrics_results def diff_logs(t1_logs, t2_logs): if t1_logs["status"] == "failed": return ( False, "Get t1 range logs failed: %s!" % t1_logs["data"]["errorMessage"], ) if t2_logs["status"] == "failed": return ( False, "Get t2 range logs failed: %s!" % t2_logs["data"]["errorMessage"], ) diff_logs_results = [] t1_logs_stats = t1_logs["data"]["logStats"] t2_logs_stats = t2_logs["data"]["logStats"] for key, value in t2_logs_stats.items(): if key == "0": continue t2_value = value[1] if key in t1_logs_stats: t1_value = t1_logs_stats[key][1] else: t1_value = 0 diff_value = float(t2_value - t1_value) / (t1_value + 1e-10) diff_logs_results.append( { "values": { "name": value[0], "t1Count": int(t1_value), "t2Count": int(t2_value), "diff": float(diff_value), "abs_diff": int(t2_value - t1_value), "level": value[2], "component": value[3], }, "subValues": None, } ) return True, diff_logs_results def generate_average_value(stats_value): results = [] for key, value in stats_value.items(): results.append("%s: %.2f" % (key, value)) return results def diff_node_max_value(stats_value): values = [] for key, value in stats_value.items(): values.append(value) values.sort() return (values[-1] - values[0]) / (values[-1] + 1e-10) def diff_node_balance(t1_metrics, t2_metrics): if t1_metrics["status"] == "failed": raise Exception("Get t1 range metrics failed: %s!" % t1_metrics["data"]["errorMessage"]) if t2_metrics["status"] == "failed": raise Exception("Get t2 range metrics failed: %s!" % t2_metrics["data"]["errorMessage"]) if not has_metrics(t1_metrics): raise Exception("T1 range metrics have no data!") if not has_metrics(t2_metrics): raise Exception("T2 range metrics have no data!") diff_node_balance_results = [] t1_stats = t1_metrics["data"]["stats"] t2_stats = t2_metrics["data"]["stats"] for stats_index in range(len(t1_stats)): if t1_stats[stats_index]["label"] == "by_instance": stats_name = t1_stats[stats_index]["metrics"] + ":" + t1_stats[stats_index]["label"] stats_value1 = t1_stats[stats_index]["value"][stats_name] stats_value2 = t2_stats[stats_index]["value"][stats_name] diff_node_balance_results.append( { "values": { "components": t1_stats[stats_index]["component"], "metrics": t1_stats[stats_index]["metrics"], "t1AverageValue": generate_average_value(stats_value1), "t1Imbalance": diff_node_max_value(stats_value1), "t2AverageValue": generate_average_value(stats_value2), "t2Imbalance": diff_node_max_value(stats_value2), }, "subValues": None, } ) return diff_node_balance_results def lambda_handler(event, context): if context.get_remaining_time_in_millis() < 60000: raise ValueError("the lambda function's timeout property should be configured at least 60s") m = __import__("app") task_class = getattr(m, "task_class")() if event.get("kwargs") is not None: event["kwargs"]["context"] = context if "resume_context" in event: ctx = event["resume_context"] print("resumed from context {ctx}") task = task_class.resume(ctx, context) else: task = task_class(**event["kwargs"]) try: result = asyncio.run( asyncio.wait_for(task.run(), context.get_remaining_time_in_millis() / 1000 - 10) ) if isinstance(result, dict) and result.get("code") == 206: return result t1_configuration_results = result[0] t1_metrics_results = result[1] t1_logs_results = result[2] t2_configuration_results = result[3] t2_metrics_results = result[4] t2_logs_results = result[5] output = { "comparisonTimeRange": [ { "values": { "t1StartTime": task.start_ts_1, "t1EndTime": task.stop_ts_1, "t2StartTime": task.start_ts_2, "t2EndTime": task.stop_ts_2, }, "subValues": None, } ], "keyMetricsComparison": diff_metrics(t1_metrics_results, t2_metrics_results, "Key"), "businessMetricsComparison": diff_metrics( t1_metrics_results, t2_metrics_results, "Workload" ), "resourceMetricsComparison": diff_metrics( t1_metrics_results, t2_metrics_results, "Resource" ), "performanceMetricsComparison": diff_perf_metrics( t1_metrics_results, t2_metrics_results ), } is_success, diff_logs_results = diff_logs(t1_logs_results, t2_logs_results) if is_success: output["logComparison"] = diff_logs_results t1_logs_err_count = t1_logs_results["data"]["errLogCount"] t2_logs_err_count = t2_logs_results["data"]["errLogCount"] logs_err_count_diff = float(t2_logs_err_count - t1_logs_err_count) / ( t1_logs_err_count + 1e-10 ) output["keyMetricsComparison"].append( { "values": { "components": "TiDB", "metrics": "Error_log_count", "label": "total", "t1AverageValue": str(t1_logs_err_count), "t2AverageValue": str(t2_logs_err_count), "diff": logs_err_count_diff, "description": { "key": "errLogCount", "variables": None, }, "grafanaLocation": None, }, "subValues": None, } ) else: if "errorMessages" in output: output["errorMessages"]["logComparison"] = diff_logs_results else: output["errorMessages"] = {} output["errorMessages"]["logComparison"] = diff_logs_results if task.org_type == "op": is_success, diff_configuration_results = diff_configuration( t1_configuration_results, t2_configuration_results ) if is_success: output["configurationComparison"] = diff_configuration_results else: if "errorMessages" in output: output["errorMessages"]["configurationComparison"] = diff_configuration_results else: output["errorMessages"] = {} output["errorMessages"]["configurationComparison"] = diff_configuration_results is_success, diff_topology_results = diff_topology_info( t1_configuration_results, t2_configuration_results ) if is_success: output["clusterTopologyInfo"] = diff_topology_results else: output["errorMessages"]["clusterTopologyInfo"] = diff_topology_results return output except Exception as e: print(e) raise e class MockContext: def get_remaining_time_in_millis(self): return 900000 if __name__ == "__main__": response = lambda_handler( { "task_id": "1", "kwargs": { "cluster": "1373933076658090925", "org_id": "4", "org_type": "cloud", "package_id_1": None, "package_id_2": None, "start_ts_1": 1658807049, "start_ts_2": 1658807200, "stop_ts_1": 1658807202, "stop_ts_2": 1658807402, }, }, MockContext(), ) print(json.dumps(response))