Merge pull request #172 from openqj/feat-metrics-extended

feat: /metrics 可观测性扩展

cloud/server.py

@@ -8224,16 +8224,33 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             done = by_status.get("completed", 0) + by_status.get("merged", 0)
             failed = by_status.get("failed", 0) + by_status.get("merge-failed", 0)
             finished = done + failed
+            # 新增可观测字段（只加键，旧键语义不动）：队列深度、
+            # 近 24h 失败数、跑超 300 秒的 running 任务数。
+            now = time.time()
+            failures_last_24h = 0
+            running_over_300s = 0
+            for row in rows:
+                if row["project_id"] not in member_ids:
+                    continue
+                if row["status"] in ("failed", "merge-failed") \
+                        and row["finished_at"] \
+                        and now - row["finished_at"] <= 86400:
+                    failures_last_24h += 1
+                if row["status"] == "running" and row["started_at"] \
+                        and now - row["started_at"] > 300:
+                    running_over_300s += 1
             try:
                 cost_rows = conn.execute(
-                    "SELECT project_id, source, total_tokens, usd_micros,"
-                    " fallback_count FROM cost_records"
+                    "SELECT project_id, model, source, total_tokens,"
+                    " usd_micros, fallback_count FROM cost_records"
                 ).fetchall()
             except (sqlite3.OperationalError, _PgOperationalError):
                 cost_rows = []
             actual_tokens = estimated_tokens = 0
             actual_usd = estimated_usd = 0
             fallback_total = 0
+            by_model = {}
+            unpriced_tokens = 0
             for row in cost_rows:
                 if row["project_id"] not in member_ids:
                     continue
@@ -8244,8 +8261,26 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     estimated_tokens += row["total_tokens"] or 0
                     estimated_usd += row["usd_micros"] or 0
                 fallback_total += row["fallback_count"] or 0
+                model = row["model"] or "unknown"
+                entry = by_model.setdefault(
+                    model, {"model": model, "total_tokens": 0,
+                            "usd_micros": 0, "priced": False})
+                entry["total_tokens"] += row["total_tokens"] or 0
+                if row["usd_micros"] is not None:
+                    entry["usd_micros"] += row["usd_micros"]
+                    entry["priced"] = True
+                else:
+                    unpriced_tokens += row["total_tokens"] or 0
+            all_tokens = actual_tokens + estimated_tokens
         finally:
             conn.close()
+        try:
+            workers = _registry_list(self.server.db_path)
+        except Exception:  # noqa: BLE001 - 观测端点不因 worker 表失败
+            workers = []
+        workers_online = sum(1 for w in workers if w.get("alive"))
+        worker_ages = [now - w["last_seen"] for w in workers
+                       if w.get("last_seen")]
         self._send_json(200, {
             "total_tasks": total,
             "by_status": by_status,
@@ -8258,6 +8293,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 "estimated_tokens": estimated_tokens,
                 "actual_usd_micros": actual_usd,
                 "estimated_usd_micros": estimated_usd,
+                "unpriced_tokens": unpriced_tokens,
+                "unpriced_share": (
+                    unpriced_tokens / all_tokens) if all_tokens else None,
+                "by_model": sorted(by_model.values(),
+                                   key=lambda e: -e["total_tokens"]),
+            },
+            "queue_depth": {
+                "queued": by_status.get("queued", 0),
+                "running": by_status.get("running", 0),
+            },
+            "failures_last_24h": failures_last_24h,
+            "running_over_300s": running_over_300s,
+            "workers": {
+                "online": workers_online,
+                "offline": len(workers) - workers_online,
+                "total": len(workers),
+                "max_last_seen_age_seconds": (
+                    max(worker_ages) if worker_ages else None),
             },
         })
 

cloud/tests/test_prod_gaps.py

@@ -287,6 +287,13 @@ class MetricsAndBackupTestCase(unittest.TestCase):
         self.assertEqual(metrics["fallback_total"], 1)
         self.assertEqual(metrics["cost"]["actual_tokens"], 15)
         self.assertEqual(metrics["cost"]["estimated_tokens"], 20)
+        # 新增观测键：队列深度、worker 聚合、分模型、未定价占比。
+        self.assertEqual(metrics["queue_depth"], {"queued": 0, "running": 0})
+        self.assertEqual(metrics["workers"]["total"], 0)
+        self.assertEqual(metrics["failures_last_24h"], 0)
+        models = {m["model"]: m for m in metrics["cost"]["by_model"]}
+        self.assertEqual(models["gpt-6.1-sol"]["total_tokens"], 35)
+        self.assertIsNotNone(metrics["cost"]["unpriced_share"])
         # 无 key 401。
         status, _ = self._req("GET", "/metrics")
         self.assertEqual(status, 401)