feat: 补全本地 K8s Ray 与 Daft 可观测性

benchmark/audio/metrics.py

@@ -5,10 +5,55 @@
 import socket
 import threading
 import time
+from urllib.parse import quote
 from pathlib import Path
 from typing import Any
 
 
+def push_run_metrics(url: str | None, summary: dict[str, Any]) -> None:
+    """将单次音频作业的聚合指标推送到 Pushgateway。
+
+    Ray Job driver 在查询完成后会退出，不能作为 Prometheus 的稳定 scrape
+    endpoint；因此仅在配置 endpoint 时以 run_id 分组推送最终结果。推送失败
+    不能影响真实数据写出或基准结论，调用方只记录告警。
+    """
+    if not url:
+        return
+
+    run_id = str(summary["run_id"])
+    state = str(summary["state"])
+    lines = [
+        "# TYPE daft_audio_run_state gauge",
+        f'daft_audio_run_state{{state="{state}"}} 1',
+        "# TYPE daft_audio_run_elapsed_seconds gauge",
+        f"daft_audio_run_elapsed_seconds {float(summary['elapsed_s'])}",
+    ]
+    for key, metric in (
+        ("rows", "daft_audio_run_rows"),
+        ("unique_doc_ids", "daft_audio_run_unique_doc_ids"),
+        ("input_bytes", "daft_audio_run_input_bytes"),
+        ("audio_seconds", "daft_audio_run_audio_seconds"),
+        ("asr_p50_ms", "daft_audio_asr_latency_p50_milliseconds"),
+        ("asr_p95_ms", "daft_audio_asr_latency_p95_milliseconds"),
+        ("llm_p50_ms", "daft_audio_llm_latency_p50_milliseconds"),
+        ("llm_p95_ms", "daft_audio_llm_latency_p95_milliseconds"),
+        ("audio_seconds_per_wall_second", "daft_audio_run_audio_seconds_per_wall_second"),
+        ("input_megabytes_per_second", "daft_audio_run_input_megabytes_per_second"),
+        ("ray_process_rss_peak_bytes", "daft_audio_ray_process_rss_peak_bytes"),
+        ("object_store_used_peak_bytes", "daft_audio_object_store_used_peak_bytes"),
+    ):
+        if summary.get(key) is not None:
+            lines.append(f"{metric} {float(summary[key])}")
+    for status, count in (summary.get("status_counts") or {}).items():
+        lines.append(f'daft_audio_run_status_rows{{status="{status}"}} {int(count)}')
+
+    import httpx
+
+    endpoint = f"{url.rstrip('/')}/metrics/job/daft_audio_smoke/run_id/{quote(run_id, safe='')}"
+    response = httpx.put(endpoint, content="\n".join(lines) + "\n", timeout=10.0)
+    response.raise_for_status()
+
+
 def _node_probe_class():
     import ray
 

benchmark/audio/pipeline.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import io
+import csv
 import json
 import os
 import signal
@@ -19,7 +20,7 @@
 from multimodal_toolkit.storage.io import daft_io_config, lance_storage_options
 
 from .config import BenchmarkConfig, runtime_metadata, write_json
-from .metrics import ResourceSampler
+from .metrics import ResourceSampler, push_run_metrics
 
 
 ASR_DTYPE = daft.DataType.struct(
@@ -323,5 +324,31 @@ def run_pipeline(cfg: BenchmarkConfig) -> dict[str, Any]:
                     "audio_seconds": sum(x or 0 for x in table.column("duration_s").to_pylist()),
                 }
             )
+            for column, prefix in (("asr_ms", "asr"), ("llm_ms", "llm")):
+                values = sorted(
+                    float(value) for value in ds.to_table(columns=[column]).column(column).to_pylist()
+                    if value is not None
+                )
+                if values:
+                    summary[f"{prefix}_p50_ms"] = values[len(values) // 2]
+                    summary[f"{prefix}_p95_ms"] = values[min(len(values) - 1, int(len(values) * 0.95))]
+            if elapsed:
+                summary["audio_seconds_per_wall_second"] = summary["audio_seconds"] / elapsed
+                summary["input_megabytes_per_second"] = summary["input_bytes"] / elapsed / 1_000_000
+        resources_path = run_dir / "resources.csv"
+        if resources_path.exists():
+            with resources_path.open(newline="") as resources_file:
+                samples = list(csv.DictReader(resources_file))
+            for field, metric in (
+                ("ray_process_rss_bytes", "ray_process_rss_peak_bytes"),
+                ("object_store_used_bytes", "object_store_used_peak_bytes"),
+            ):
+                values = [float(row[field]) for row in samples if row.get(field)]
+                if values:
+                    summary[metric] = max(values)
         write_json(run_dir / "summary.json", summary)
+        try:
+            push_run_metrics(os.getenv("BENCH_PROMETHEUS_PUSHGATEWAY_URL"), summary)
+        except Exception as exc:
+            print(f"[warn] failed to push Prometheus metrics: {type(exc).__name__}: {exc}")
     return summary

deploy/k8s/grafana.yaml

@@ -0,0 +1,134 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: grafana-credentials
+  namespace: daft-ray
+type: Opaque
+stringData:
+  GF_SECURITY_ADMIN_USER: admin
+  GF_SECURITY_ADMIN_PASSWORD: admin
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: grafana-provisioning
+  namespace: daft-ray
+data:
+  datasource.yaml: |
+    apiVersion: 1
+    datasources:
+      - name: Prometheus
+        uid: prometheus
+        type: prometheus
+        access: proxy
+        url: http://prometheus:9090
+        isDefault: true
+  dashboard.yaml: |
+    apiVersion: 1
+    providers:
+      - name: Daft Ray
+        type: file
+        options:
+          path: /var/lib/grafana/dashboards
+  daft-ray.json: |
+    {"annotations":{"list":[]},"editable":true,"panels":[{"type":"stat","title":"Ray 平均 CPU 使用率 (%)","gridPos":{"h":5,"w":4,"x":0,"y":0},"targets":[{"expr":"avg(ray_node_cpu_utilization)","refId":"A"}]},{"type":"stat","title":"Ray 运行中 Job","gridPos":{"h":5,"w":4,"x":4,"y":0},"targets":[{"expr":"sum(ray_running_jobs)","refId":"A"}]},{"type":"stat","title":"Ray 已完成 Job","gridPos":{"h":5,"w":4,"x":8,"y":0},"targets":[{"expr":"sum(ray_finished_jobs_total)","refId":"A"}]},{"type":"stat","title":"Daft 最新作业耗时 (s)","gridPos":{"h":5,"w":4,"x":12,"y":0},"targets":[{"expr":"daft_audio_run_elapsed_seconds","refId":"A"}]},{"type":"stat","title":"Daft 音频实时倍率","gridPos":{"h":5,"w":4,"x":16,"y":0},"targets":[{"expr":"daft_audio_run_audio_seconds_per_wall_second","refId":"A"}]},{"type":"stat","title":"Daft 输入吞吐 (MB/s)","gridPos":{"h":5,"w":4,"x":20,"y":0},"targets":[{"expr":"daft_audio_run_input_megabytes_per_second","refId":"A"}]},{"type":"timeseries","title":"Ray 节点内存使用率 (%)","gridPos":{"h":8,"w":12,"x":0,"y":5},"targets":[{"expr":"100 * ray_node_mem_used / ray_node_mem_total","refId":"A"}]},{"type":"timeseries","title":"Ray Object Store 已用内存","gridPos":{"h":8,"w":12,"x":12,"y":5},"targets":[{"expr":"sum(ray_object_store_used_memory)","refId":"A"},{"expr":"sum(ray_object_store_available_memory)","refId":"B"}]},{"type":"timeseries","title":"Daft ASR/LLM P50/P95 延迟 (ms)","gridPos":{"h":8,"w":12,"x":0,"y":13},"targets":[{"expr":"daft_audio_asr_latency_p50_milliseconds","refId":"A"},{"expr":"daft_audio_asr_latency_p95_milliseconds","refId":"B"},{"expr":"daft_audio_llm_latency_p50_milliseconds","refId":"C"},{"expr":"daft_audio_llm_latency_p95_milliseconds","refId":"D"}]},{"type":"timeseries","title":"Daft 状态行数","gridPos":{"h":8,"w":12,"x":12,"y":13},"targets":[{"expr":"daft_audio_run_status_rows","refId":"A"}]},{"type":"timeseries","title":"Daft Ray 进程 RSS 峰值","gridPos":{"h":8,"w":12,"x":0,"y":21},"targets":[{"expr":"daft_audio_ray_process_rss_peak_bytes","refId":"A"}]},{"type":"timeseries","title":"Daft Object Store 峰值","gridPos":{"h":8,"w":12,"x":12,"y":21},"targets":[{"expr":"daft_audio_object_store_used_peak_bytes","refId":"A"}]}],"schemaVersion":41,"title":"Daft 与 Ray 本地运行概览","uid":"daft-ray-local","version":2}
+  ray-runtime.json: |
+    {
+      "annotations": {"list": []},
+      "description": "Ray 集群运行态与进程级资源诊断。通过顶部筛选器缩小到指定节点或 Ray 组件；红色阈值表示需要优先排查的资源或调度异常。",
+      "editable": true,
+      "panels": [
+        {"type":"stat","title":"活跃 Ray 节点","gridPos":{"h":5,"w":4,"x":0,"y":0},"targets":[{"expr":"sum(ray_cluster_active_nodes)","refId":"A"}]},
+        {"type":"stat","title":"运行中 Job","gridPos":{"h":5,"w":4,"x":4,"y":0},"targets":[{"expr":"sum(ray_running_jobs)","refId":"A"}]},
+        {"type":"stat","title":"Actor 总数","gridPos":{"h":5,"w":4,"x":8,"y":0},"targets":[{"expr":"sum(ray_actors)","refId":"A"}]},
+        {"type":"stat","title":"不可调度任务","gridPos":{"h":5,"w":4,"x":12,"y":0},"targets":[{"expr":"sum(ray_scheduler_unscheduleable_tasks)","refId":"A"}]},
+        {"type":"stat","title":"对象 Store 本地对象","gridPos":{"h":5,"w":4,"x":16,"y":0},"targets":[{"expr":"sum(ray_object_store_num_local_objects)","refId":"A"}]},
+        {"type":"stat","title":"Worker 启动失败增量","gridPos":{"h":5,"w":4,"x":20,"y":0},"targets":[{"expr":"sum(increase(ray_scheduler_failed_worker_startup_total[$__rate_interval]))","refId":"A"}]},
+        {"type":"timeseries","title":"Ray 节点 CPU 使用率","gridPos":{"h":8,"w":12,"x":0,"y":5},"fieldConfig":{"defaults":{"unit":"percent","thresholds":{"mode":"absolute","steps":[{"color":"green","value":null},{"color":"orange","value":70},{"color":"red","value":90}]}},"overrides":[]},"targets":[{"expr":"ray_node_cpu_utilization{ip=~\"$node\"}","legendFormat":"{{ip}}","refId":"A"}]},
+        {"type":"timeseries","title":"Ray 节点内存使用率","gridPos":{"h":8,"w":12,"x":12,"y":5},"fieldConfig":{"defaults":{"unit":"percent","thresholds":{"mode":"absolute","steps":[{"color":"green","value":null},{"color":"orange","value":75},{"color":"red","value":90}]}},"overrides":[]},"targets":[{"expr":"100 * ray_node_mem_used{ip=~\"$node\"} / ray_node_mem_total{ip=~\"$node\"}","legendFormat":"{{ip}}","refId":"A"}]},
+        {"type":"timeseries","title":"Ray 进程 CPU 使用率","gridPos":{"h":9,"w":12,"x":0,"y":13},"fieldConfig":{"defaults":{"unit":"percent","thresholds":{"mode":"absolute","steps":[{"color":"green","value":null},{"color":"orange","value":70},{"color":"red","value":90}]}},"overrides":[]},"targets":[{"expr":"ray_component_cpu_percentage{ip=~\"$node\",Component=~\"$component\"}","legendFormat":"{{ip}} {{Component}} pid={{pid}}","refId":"A"}]},
+        {"type":"timeseries","title":"Ray 进程 RSS 内存","gridPos":{"h":9,"w":12,"x":12,"y":13},"fieldConfig":{"defaults":{"unit":"decmbytes"},"overrides":[]},"targets":[{"expr":"ray_component_rss_mb{ip=~\"$node\",Component=~\"$component\"}","legendFormat":"{{ip}} {{Component}} pid={{pid}}","refId":"A"}]},
+        {"type":"timeseries","title":"Ray 进程 USS 内存","gridPos":{"h":9,"w":12,"x":0,"y":22},"fieldConfig":{"defaults":{"unit":"decmbytes"},"overrides":[]},"targets":[{"expr":"ray_component_uss_mb{ip=~\"$node\",Component=~\"$component\"}","legendFormat":"{{ip}} {{Component}} pid={{pid}}","refId":"A"}]},
+        {"type":"timeseries","title":"Ray 进程共享内存","gridPos":{"h":9,"w":12,"x":12,"y":22},"fieldConfig":{"defaults":{"unit":"bytes"},"overrides":[]},"targets":[{"expr":"ray_component_mem_shared_bytes{ip=~\"$node\",Component=~\"$component\"}","legendFormat":"{{ip}} {{Component}} pid={{pid}}","refId":"A"}]},
+        {"type":"timeseries","title":"Ray 进程文件句柄数","gridPos":{"h":8,"w":12,"x":0,"y":31},"fieldConfig":{"defaults":{"unit":"none"},"overrides":[]},"targets":[{"expr":"ray_component_num_fds{ip=~\"$node\",Component=~\"$component\"}","legendFormat":"{{ip}} {{Component}} pid={{pid}}","refId":"A"}]},
+        {"type":"timeseries","title":"Object Store 内存 (Bytes)","gridPos":{"h":8,"w":12,"x":12,"y":31},"targets":[{"expr":"sum(ray_object_store_used_memory)","legendFormat":"used","refId":"A"},{"expr":"sum(ray_object_store_available_memory)","legendFormat":"available","refId":"B"},{"expr":"sum(ray_object_store_fallback_memory)","legendFormat":"fallback","refId":"C"}]},
+        {"type":"timeseries","title":"Ray 网络吞吐 (Bytes/s)","gridPos":{"h":8,"w":12,"x":0,"y":39},"targets":[{"expr":"ray_node_network_receive_speed","legendFormat":"{{ip}} receive","refId":"A"},{"expr":"ray_node_network_send_speed","legendFormat":"{{ip}} send","refId":"B"}]},
+        {"type":"timeseries","title":"Ray 磁盘使用率 (%)","gridPos":{"h":8,"w":12,"x":12,"y":39},"targets":[{"expr":"ray_node_disk_utilization_percentage","legendFormat":"{{ip}}","refId":"A"}]},
+        {"type":"timeseries","title":"调度任务与资源","gridPos":{"h":8,"w":12,"x":0,"y":47},"targets":[{"expr":"sum(ray_scheduler_tasks)","legendFormat":"scheduled tasks","refId":"A"},{"expr":"sum(ray_scheduler_unscheduleable_tasks)","legendFormat":"unschedulable tasks","refId":"B"},{"expr":"sum(ray_resources)","legendFormat":"resources","refId":"C"}]},
+        {"type":"timeseries","title":"对象传输与拉取请求","gridPos":{"h":8,"w":12,"x":12,"y":47},"targets":[{"expr":"sum(ray_object_manager_num_pull_requests)","legendFormat":"pull requests","refId":"A"},{"expr":"sum(ray_pull_manager_requests)","legendFormat":"pull manager requests","refId":"B"},{"expr":"sum(ray_pull_manager_active_bundles)","legendFormat":"active bundles","refId":"C"}]},
+        {"type":"timeseries","title":"GCS 任务事件","gridPos":{"h":8,"w":12,"x":0,"y":55},"targets":[{"expr":"sum(ray_gcs_task_manager_task_events_reported)","legendFormat":"reported","refId":"A"},{"expr":"sum(ray_gcs_task_manager_task_events_stored)","legendFormat":"stored","refId":"B"},{"expr":"sum(ray_gcs_task_manager_task_events_dropped)","legendFormat":"dropped","refId":"C"}]},
+        {"type":"timeseries","title":"Ray 内部进程与 I/O Event Loop 延迟","gridPos":{"h":8,"w":12,"x":12,"y":55},"targets":[{"expr":"sum(ray_internal_num_processes_started)","legendFormat":"processes started","refId":"A"},{"expr":"ray_io_context_event_loop_lag_ms","legendFormat":"{{ip}} event loop lag ms","refId":"B"}]}
+      ],
+      "refresh": "10s",
+      "schemaVersion": 41,
+      "tags": ["Ray", "Kubernetes", "进程诊断"],
+      "templating": {
+        "list": [
+          {"name":"node","label":"Ray 节点","type":"query","datasource":{"uid":"prometheus"},"query":"label_values(ray_node_cpu_utilization, ip)","includeAll":true,"allValue":".*","multi":true,"current":{"text":"All","value":"$__all"}},
+          {"name":"component","label":"Ray 组件","type":"query","datasource":{"uid":"prometheus"},"query":"label_values(ray_component_cpu_percentage, Component)","includeAll":true,"allValue":".*","multi":true,"current":{"text":"All","value":"$__all"}}
+        ]
+      },
+      "time": {"from":"now-1h","to":"now"},
+      "title": "Ray 运行时与进程诊断",
+      "uid": "ray-runtime-local",
+      "version": 1
+    }
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: grafana
+  namespace: daft-ray
+spec:
+  replicas: 1
+  selector:
+    matchLabels: {app: grafana}
+  template:
+    metadata:
+      labels: {app: grafana}
+    spec:
+      containers:
+        - name: grafana
+          image: grafana/grafana:12.0.2
+          envFrom:
+            - secretRef: {name: grafana-credentials}
+          env:
+            # Grafana 只经本机 port-forward 暴露；匿名用户仅能查看预置看板。
+            - name: GF_AUTH_ANONYMOUS_ENABLED
+              value: "true"
+            - name: GF_AUTH_ANONYMOUS_ORG_ROLE
+              value: Viewer
+          ports:
+            - containerPort: 3000
+              name: web
+          resources:
+            requests: {cpu: 100m, memory: 256Mi}
+            limits: {cpu: 500m, memory: 512Mi}
+          volumeMounts:
+            - name: provisioning
+              mountPath: /etc/grafana/provisioning/datasources/datasource.yaml
+              subPath: datasource.yaml
+            - name: provisioning
+              mountPath: /etc/grafana/provisioning/dashboards/dashboard.yaml
+              subPath: dashboard.yaml
+            - name: provisioning
+              mountPath: /var/lib/grafana/dashboards/daft-ray.json
+              subPath: daft-ray.json
+            - name: provisioning
+              mountPath: /var/lib/grafana/dashboards/ray-runtime.json
+              subPath: ray-runtime.json
+      volumes:
+        - name: provisioning
+          configMap: {name: grafana-provisioning}
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: grafana
+  namespace: daft-ray
+spec:
+  selector: {app: grafana}
+  ports:
+    - name: web
+      port: 3000
+      targetPort: web

deploy/k8s/prometheus.yaml

@@ -10,6 +10,9 @@ data:
       scrape_interval: 5s
       evaluation_interval: 5s
     scrape_configs:
+      - job_name: daft-audio-pushgateway
+        static_configs:
+          - targets: [pushgateway:9091]
       - job_name: ray
         kubernetes_sd_configs:
           # Pod discovery 配合 Ray 节点标签，避免 Endpoints 元数据在不同

deploy/k8s/pushgateway.yaml

@@ -0,0 +1,34 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: pushgateway
+  namespace: daft-ray
+spec:
+  replicas: 1
+  selector:
+    matchLabels: {app: pushgateway}
+  template:
+    metadata:
+      labels: {app: pushgateway}
+    spec:
+      containers:
+        - name: pushgateway
+          image: prom/pushgateway:v1.11.1
+          ports:
+            - containerPort: 9091
+              name: web
+          resources:
+            requests: {cpu: 50m, memory: 64Mi}
+            limits: {cpu: 200m, memory: 128Mi}
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: pushgateway
+  namespace: daft-ray
+spec:
+  selector: {app: pushgateway}
+  ports:
+    - name: web
+      port: 9091
+      targetPort: web

deploy/k8s/raycluster.yaml

@@ -30,6 +30,8 @@ spec:
                 value: minio,mock-llm,127.0.0.1,localhost
               - name: no_proxy
                 value: minio,mock-llm,127.0.0.1,localhost
+              - name: BENCH_PROMETHEUS_PUSHGATEWAY_URL
+                value: http://pushgateway:9091
             envFrom:
               - secretRef:
                   name: minio-credentials

docs/k8s-ray-local.md

@@ -42,6 +42,7 @@ kubectl --context daft-ray -n daft-ray port-forward service/prometheus 9090:9090
 
 - Ray Dashboard：<http://127.0.0.1:8265>
 - Prometheus targets：<http://127.0.0.1:9090/targets>
+- Grafana：<http://127.0.0.1:3000>（本地开发账号 `admin` / `admin`）
 
 所有 Ray target 应为 `UP`。可从以下 PromQL 起步；实际可用指标名以 target 的 `/metrics`
 为准，因为会随 Ray 版本和组件变化：
@@ -52,6 +53,10 @@ ray_object_store_memory
 ray_node_cpu_utilization
 ```
 
+Grafana 内置“Daft 与 Ray 本地运行概览”看板：Ray 的 CPU 与 object store 来自 5 秒抓取；
+Daft 音频作业在完成时通过 Pushgateway 写入总耗时、行数、状态、ASR/LLM P95 延迟。指标仅
+保留当前环境生命周期，重建 Prometheus 或 Pushgateway 后会清空。
+
 ## 运行真实音频 smoke
 
 ```bash

scripts/k8s/run-audio-smoke.sh

@@ -20,6 +20,7 @@ common_env=(
   "MINIO_ROOT_USER=minioadmin"
   "MINIO_ROOT_PASSWORD=minioadmin"
   "MINIO_REGION=us-east-1"
+  "BENCH_PROMETHEUS_PUSHGATEWAY_URL=http://pushgateway:9091"
   "NO_PROXY=minio,mock-llm,127.0.0.1,localhost"
   "no_proxy=minio,mock-llm,127.0.0.1,localhost"
 )

scripts/k8s/status.sh

@@ -13,6 +13,10 @@ Prometheus:
   kubectl --context daft-ray -n daft-ray port-forward service/prometheus 9090:9090
   open http://127.0.0.1:9090/targets
 
+Grafana:
+  kubectl --context daft-ray -n daft-ray port-forward service/grafana 3000:3000
+  open http://127.0.0.1:3000
+
 MinIO Console:
   kubectl --context daft-ray -n daft-ray port-forward service/minio 9001:9001
   open http://127.0.0.1:9001