Grafana数据管道：监控面板设计

import requestsimport time def query_prometheus(prom_url, promql_expr, step="60s"):    """向Prometheus发送范围查询并返回时间序列数据"""    now = int(time.time())    resp = requests.get(        f"{prom_url}/api/v1/query_range",        params={            "query": promql_expr,            "start": now - 3600,            "end": now,            "step": step        },        timeout=10    )    resp.raise_for_status()    data = resp.json()    return data["data"]["result"] # 查询GPU利用率的平均值results = query_prometheus(    "http://prometheus:9090",    'avg(gpu_utilization{job="ml-training"})')for series in results:    metric = series["metric"]    values = series["values"]    print(f"实例: {metric}, 数据点数: {len(values)}")

def recommend_panel_type(data_characteristics):    """根据数据特征推荐合适的Grafana面板类型"""    field_types = data_characteristics.get("field_types", [])    has_time = "timestamp" in field_types    num_series = data_characteristics.get("num_series", 1)    is_distribution = data_characteristics.get("is_distribution", False)     if is_distribution:        return {"panel": "heatmap", "reason": "数据为分布类型，热力图最适合展示"}    if has_time and num_series > 1:        return {"panel": "time_series", "reason": "多序列时序数据，推荐时序图"}    if has_time and num_series == 1:        return {"panel": "time_series", "reason": "单序列时序数据，推荐时序图"}    if not has_time and num_series <= 4:        return {"panel": "stat", "reason": "少量指标无时间维度，推荐统计值面板"}    return {"panel": "table", "reason": "通用表格展示"} # 测试推荐逻辑result = recommend_panel_type({    "field_types": ["timestamp", "value"],    "num_series": 5,    "is_distribution": False})print(f"推荐面板: {result['panel']}, 原因: {result['reason']}")

class AlertRule:    """模拟Grafana告警规则引擎"""    def __init__(self, name, metric, threshold, operator="gt", for_duration=300):        self.name = name        self.metric = metric        self.threshold = threshold        self.operator = operator        self.for_duration = for_duration        self.triggered_since = None     def evaluate(self, current_value, timestamp):        """评估当前值是否触发告警条件"""        is_breach = False        if self.operator == "gt" and current_value > self.threshold:            is_breach = True        elif self.operator == "lt" and current_value < self.threshold:            is_breach = True         if is_breach:            if self.triggered_since is None:                self.triggered_since = timestamp            elapsed = timestamp - self.triggered_since            if elapsed >= self.for_duration:                return "firing"            return "pending"        else:            self.triggered_since = None            return "ok" rule = AlertRule("GPU温度过高", "gpu_temp", threshold=85, for_duration=120)print(rule.evaluate(88, 1000))  # pendingprint(rule.evaluate(90, 1130))  # firing

import json def generate_grafana_dashboard(title, panels_config):    """根据配置自动生成Grafana面板JSON定义"""    dashboard = {        "title": title,        "panels": [],        "time": {"from": "now-6h", "to": "now"},        "refresh": "30s"    }    for idx, pc in enumerate(panels_config):        panel = {            "id": idx + 1,            "type": pc["panel_type"],            "title": pc["title"],            "gridPos": {"x": (idx % 2) * 12, "y": (idx // 2) * 8, "w": 12, "h": 8},            "targets": [{                "expr": pc["promql"],                "refId": "A"            }],            "fieldConfig": pc.get("field_config", {})        }        dashboard["panels"].append(panel)    return dashboard panels = [    {"panel_type": "timeseries", "title": "GPU利用率", "promql": "gpu_utilization"},    {"panel_type": "stat", "title": "训练吞吐量", "promql": "training_throughput"}]dash = generate_grafana_dashboard("AI训练监控", panels)print(json.dumps(dash, ensure_ascii=False, indent=2))

def validate_promql_semantics(promql_expr, known_metrics):    """校验PromQL表达式的语义正确性"""    import re    # 提取指标名称    metrics = re.findall(r'([a-zA-Z_:][a-zA-Z0-9_:]*)', promql_expr)    # 过滤掉PromQL函数和关键字    promql_keywords = {'sum', 'avg', 'rate', 'histogram_quantile', 'by', 'without'}    user_metrics = [m for m in metrics if m not in promql_keywords]     validation_result = {"valid": True, "errors": []}    for metric in user_metrics:        if metric not in known_metrics:            validation_result["valid"] = False            validation_result["errors"].append(f"未知指标: {metric}")    return validation_result known = {"gpu_utilization", "training_loss", "inference_latency_ms"}result = validate_promql_semantics("avg(gpu_utilization) / cpu_count", known)print(f"校验结果: {result}")

import jsonimport randomimport time def simulate_metrics(num_points=60):    """模拟生成AI训练集群的监控指标数据"""    timestamps = [int(time.time()) - (num_points - i) * 60 for i in range(num_points)]    gpu_utils = [60 + random.uniform(-10, 25) for _ in range(num_points)]    train_loss = [2.5 * (0.95 ** i) + random.uniform(-0.05, 0.05) for i in range(num_points)]    return timestamps, gpu_utils, train_loss def detect_anomaly(values, threshold_factor=2.0):    """基于均值和标准差检测异常点"""    mean_val = sum(values) / len(values)    std_val = (sum((v - mean_val) ** 2 for v in values) / len(values)) ** 0.5    upper = mean_val + threshold_factor * std_val    lower = mean_val - threshold_factor * std_val    anomalies = [(i, v) for i, v in enumerate(values) if v > upper or v < lower]    return anomalies, mean_val, std_val timestamps, gpu_utils, train_loss = simulate_metrics(60)anomalies, mean_gpu, std_gpu = detect_anomaly(gpu_utils)print(f"GPU利用率均值: {mean_gpu:.1f}%, 标准差: {std_gpu:.1f}%")print(f"检测到 {len(anomalies)} 个异常点")for idx, val in anomalies:    print(f"  时间点 {idx}: {val:.1f}%") dashboard_config = {    "title": "AI训练监控面板",    "panels": [        {"type": "timeseries", "metric": "gpu_utilization", "unit": "percent"},        {"type": "timeseries", "metric": "training_loss", "unit": "loss"}    ],    "alert_rules": [        {"metric": "gpu_utilization", "threshold": 95, "severity": "warning"}    ]}print(json.dumps(dashboard_config, ensure_ascii=False, indent=2))