fix: 治理 Prometheus 指标基数爆炸和内存缓存无界增长

- 删除高基数标签 key_id/model，移除未使用的指标 concurrency_slots_in_use/streaming_request_duration_seconds
- HTTP 客户端池：命名客户端添加 LRU 淘汰上限，tunnel 客户端添加 LRU 淘汰和 last_used_time 追踪
- ResilienceManager：last_errors 改用 deque(maxlen)，error_stats/circuit_breakers 添加上限淘汰
- HealthMonitor：_circuit_history 改用 deque(maxlen)
- ProviderHealthTracker：清理已无记录的过期 key
- PrometheusPlugin：动态指标数量添加上限，超限后拒绝创建
- 中间件使用路由模板替代实际路径，防止动态路径段导致标签爆炸

src/clients/http_client.py

@@ -48,12 +48,14 @@ class HTTPClientPool:
     _instance: HTTPClientPool | None = None
     _default_client: httpx.AsyncClient | None = None
     _clients: dict[str, httpx.AsyncClient] = {}
+    _max_named_clients: int = 20
     # 代理客户端缓存：{cache_key: (client, last_used_time)}
     _proxy_clients: dict[str, tuple[httpx.AsyncClient, float]] = {}
     # 代理客户端缓存上限（避免内存泄漏）
     _max_proxy_clients: int = 50
-    # Tunnel 客户端缓存：{node_id: client}
-    _tunnel_clients: dict[str, httpx.AsyncClient] = {}
+    # Tunnel 客户端缓存：{node_id: (client, last_used_time)}
+    _tunnel_clients: dict[str, tuple[httpx.AsyncClient, float]] = {}
+    _max_tunnel_clients: int = 30
     # 后台清理任务引用集合（防止被 GC 回收）
     _background_tasks: set[asyncio.Task[None]] = set()
 
@@ -143,23 +145,37 @@ class HTTPClientPool:
             name: 客户端标识符
             **kwargs: httpx.AsyncClient的配置参数
         """
-        if name not in cls._clients:
-            # 合并默认配置和自定义配置
-            default_config = {
-                "http2": config.enable_http2,
-                "verify": get_ssl_context(),
-                "timeout": httpx.Timeout(
-                    connect=config.http_connect_timeout,
-                    read=config.http_read_timeout,
-                    write=config.http_write_timeout,
-                    pool=config.http_pool_timeout,
-                ),
-                "follow_redirects": True,
-            }
-            default_config.update(kwargs)
+        if name in cls._clients:
+            # 命中缓存：移到末尾以维护 LRU 顺序
+            cls._clients[name] = cls._clients.pop(name)
+            return cls._clients[name]
 
-            cls._clients[name] = httpx.AsyncClient(**default_config)  # type: ignore[arg-type]
-            logger.debug("创建命名HTTP客户端: {}", name)
+        # 淘汰最久未使用的客户端（dict 头部即 LRU）
+        if len(cls._clients) >= cls._max_named_clients:
+            oldest_name = next(iter(cls._clients))
+            old_client = cls._clients.pop(oldest_name)
+            try:
+                asyncio.get_running_loop().create_task(old_client.aclose())
+            except RuntimeError:
+                pass
+            logger.debug("淘汰命名HTTP客户端: {}", oldest_name)
+
+        # 合并默认配置和自定义配置
+        default_config = {
+            "http2": config.enable_http2,
+            "verify": get_ssl_context(),
+            "timeout": httpx.Timeout(
+                connect=config.http_connect_timeout,
+                read=config.http_read_timeout,
+                write=config.http_write_timeout,
+                pool=config.http_pool_timeout,
+            ),
+            "follow_redirects": True,
+        }
+        default_config.update(kwargs)
+
+        cls._clients[name] = httpx.AsyncClient(**default_config)  # type: ignore[arg-type]
+        logger.debug("创建命名HTTP客户端: {}", name)
 
         return cls._clients[name]
 
@@ -357,7 +373,7 @@ class HTTPClientPool:
         cls._proxy_clients.clear()
 
         # 关闭 tunnel 客户端缓存
-        for nid, client in cls._tunnel_clients.items():
+        for nid, (client, _) in cls._tunnel_clients.items():
             try:
                 await client.aclose()
                 logger.debug("tunnel 客户端已关闭: {}", nid)
@@ -524,9 +540,10 @@ class HTTPClientPool:
                 return False
             lock = cls._get_proxy_clients_lock()
             async with lock:
-                client = cls._tunnel_clients.pop(node_id, None)
-            if client is None:
+                entry = cls._tunnel_clients.pop(node_id, None)
+            if entry is None:
                 return False
+            client, _ = entry
             try:
                 await client.aclose()
             except Exception as exc:
@@ -635,13 +652,27 @@ class HTTPClientPool:
         # 非流式请求：复用缓存的 client（加锁与 proxy_clients 保持一致）
         lock = cls._get_proxy_clients_lock()
         async with lock:
-            existing = cls._tunnel_clients.get(node_id)
-            if existing and not existing.is_closed:
-                return existing
+            entry = cls._tunnel_clients.get(node_id)
+            if entry is not None:
+                existing, _ = entry
+                if not existing.is_closed:
+                    cls._tunnel_clients[node_id] = (existing, time.time())
+                    return existing
+                del cls._tunnel_clients[node_id]
+
+            # 淘汰最久未使用的 tunnel 客户端
+            if len(cls._tunnel_clients) >= cls._max_tunnel_clients:
+                oldest_nid = min(cls._tunnel_clients, key=lambda k: cls._tunnel_clients[k][1])
+                old_client, _ = cls._tunnel_clients.pop(oldest_nid)
+                try:
+                    await old_client.aclose()
+                except Exception:
+                    pass
+                logger.debug("淘汰 tunnel 客户端: {}", oldest_nid)
 
             transport = create_tunnel_transport(node_id, timeout=timeout_secs or 60.0)
             client = httpx.AsyncClient(transport=transport, timeout=t)
-            cls._tunnel_clients[node_id] = client
+            cls._tunnel_clients[node_id] = (client, time.time())
             return client
 
     @classmethod

src/core/metrics.py

@@ -4,11 +4,11 @@ Prometheus metrics for monitoring
 
 from prometheus_client import Counter, Gauge, Histogram
 
-# 并发槽位占用时长分布
+# 并发槽位占用时长分布（按异常类型聚合，不按 key_id 拆分）
 concurrency_slot_duration_seconds = Histogram(
     "concurrency_slot_duration_seconds",
     "Duration of concurrency slot occupation in seconds",
-    ["key_id", "exception"],
+    ["exception"],
     buckets=[0.1, 0.5, 1, 5, 10, 30, 60, 120, 300, 600],  # 0.1s 到 10 分钟
 )
 
@@ -16,20 +16,7 @@ concurrency_slot_duration_seconds = Histogram(
 concurrency_slot_release_total = Counter(
     "concurrency_slot_release_total",
     "Total number of concurrency slot releases",
-    ["key_id", "exception"],
-)
-
-# 当前并发槽位使用数
-concurrency_slots_in_use = Gauge(
-    "concurrency_slots_in_use", "Current number of concurrency slots in use", ["key_id"]
-)
-
-# 流式请求时长分布
-streaming_request_duration_seconds = Histogram(
-    "streaming_request_duration_seconds",
-    "Duration of streaming requests in seconds",
-    ["key_id", "status"],
-    buckets=[1, 5, 10, 30, 60, 120, 300, 600, 1800],  # 1s 到 30 分钟
+    ["exception"],
 )
 
 # 请求总数（按类型）
@@ -118,5 +105,5 @@ billing_invariant_violation_total = Counter(
 antigravity_degradation_total = Counter(
     "aether_antigravity_degradation_total",
     "Count of Antigravity signature degradation (rectification) events",
-    ["stage", "model"],
+    ["stage"],
 )

src/core/provider_health.py

@@ -110,6 +110,16 @@ class ProviderHealthTracker:
             if all(current_time - t > self.recovery_time for t in self.successes[provider_name]):
                 self.priority_adjustments[provider_name] = 0
 
+        # 清理已无记录且无优先级调整的 key，防止 dict 无限增长
+        if (
+            not self.failures[provider_name]
+            and not self.successes[provider_name]
+            and self.priority_adjustments.get(provider_name, 0) == 0
+        ):
+            self.failures.pop(provider_name, None)
+            self.successes.pop(provider_name, None)
+            self.priority_adjustments.pop(provider_name, None)
+
     def _get_status_label(self, failure_rate: float, recent_failures: int) -> str:
         """根据失败率返回状态标签"""
         if recent_failures >= self.failure_threshold:

src/core/resilience.py

@@ -11,6 +11,7 @@ import threading
 import time
 import traceback
 import uuid
+from collections import deque
 from collections.abc import Callable
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
@@ -115,11 +116,15 @@ class CircuitBreaker:
 class ResilienceManager:
     """系统韧性管理器"""
 
+    _MAX_ERROR_STATS = 500
+    _MAX_CIRCUIT_BREAKERS = 200
+    _MAX_LAST_ERRORS = 100
+
     def __init__(self) -> None:
         self.error_patterns: list[ErrorPattern] = []
         self.circuit_breakers: dict[str, CircuitBreaker] = {}
         self.error_stats: dict[str, int] = {}
-        self.last_errors: list[dict[str, Any]] = []
+        self.last_errors: deque[dict[str, Any]] = deque(maxlen=self._MAX_LAST_ERRORS)
         self._setup_default_patterns()
 
     def _setup_default_patterns(self) -> None:
@@ -203,6 +208,11 @@ class ResilienceManager:
     def get_circuit_breaker(self, key: str) -> CircuitBreaker:
         """获取或创建熔断器"""
         if key not in self.circuit_breakers:
+            # 淘汰已恢复的旧熔断器，防止无界增长
+            if len(self.circuit_breakers) >= self._MAX_CIRCUIT_BREAKERS:
+                closed_keys = [k for k, cb in self.circuit_breakers.items() if cb.state == "closed"]
+                for k in closed_keys:
+                    del self.circuit_breakers[k]
             self.circuit_breakers[key] = CircuitBreaker()
         return self.circuit_breakers[key]
 
@@ -226,13 +236,18 @@ class ResilienceManager:
         }
 
         self.last_errors.append(error_info)
-        # 只保留最近100个错误
-        if len(self.last_errors) > 100:
-            self.last_errors.pop(0)
 
-        # 更新错误统计
+        # 更新错误统计（超上限时淘汰计数最低的条目）
         error_key = f"{type(error).__name__}:{operation}"
         self.error_stats[error_key] = self.error_stats.get(error_key, 0) + 1
+        if len(self.error_stats) > self._MAX_ERROR_STATS:
+            min_key = min(
+                (k for k in self.error_stats if k != error_key),
+                key=lambda k: self.error_stats[k],
+                default=None,
+            )
+            if min_key is not None:
+                del self.error_stats[min_key]
 
         # 查找匹配的错误处理模式
         pattern = self._find_matching_pattern(error)

src/middleware/plugin_middleware.py

@@ -439,9 +439,13 @@ class PluginMiddleware:
         monitor_plugin = self.plugin_manager.get_plugin("monitor")
         if monitor_plugin and monitor_plugin.enabled:
             try:
+                # 使用路由模板而非实际路径，避免动态段导致 Prometheus 标签基数爆炸
+                route = request.scope.get("route")
+                endpoint_label = route.path if route and hasattr(route, "path") else "unknown"
+
                 monitor_labels = {
                     "method": request.method,
-                    "endpoint": request.url.path,
+                    "endpoint": endpoint_label,
                     "status": str(status_code),
                     "status_class": f"{status_code // 100}xx",
                 }

src/plugins/monitor/prometheus.py

@@ -39,6 +39,7 @@ class PrometheusPlugin(MonitorPlugin):
 
         # 指标注册表
         self._metrics: dict[str, Any] = {}
+        self._max_dynamic_metrics: int = 100  # 动态创建的指标数上限
         self._buffer: list[Metric] = []
         self._lock = asyncio.Lock()
         self._flush_task: asyncio.Task | None = None  # 跟踪后台任务
@@ -140,9 +141,12 @@ class PrometheusPlugin(MonitorPlugin):
 
     def _get_or_create_metric(
         self, name: str, metric_type: MetricType, labels: list[str] | None = None
-    ) -> Any:
-        """获取或创建指标"""
+    ) -> Any | None:
+        """获取或创建指标，超过上限后拒绝创建新指标"""
         if name not in self._metrics:
+            if len(self._metrics) >= self._max_dynamic_metrics:
+                logger.warning("Prometheus dynamic metrics limit reached, dropping: {}", name)
+                return None
             labels = labels or []
             if metric_type == MetricType.COUNTER:
                 self._metrics[name] = Counter(name, f"Auto-created counter {name}", labels)
@@ -153,7 +157,7 @@ class PrometheusPlugin(MonitorPlugin):
             elif metric_type == MetricType.SUMMARY:
                 self._metrics[name] = Summary(name, f"Auto-created summary {name}", labels)
 
-        return self._metrics[name]
+        return self._metrics.get(name)
 
     async def record_metric(self, metric: Metric) -> None:
         """记录单个指标"""
@@ -190,6 +194,8 @@ class PrometheusPlugin(MonitorPlugin):
                 # 创建新的计数器
                 label_names = list(labels.keys()) if labels else []
                 metric = self._get_or_create_metric(name, MetricType.COUNTER, label_names)
+                if metric is None:
+                    return
                 if labels:
                     metric.labels(**labels).inc(value)
                 else:
@@ -212,6 +218,8 @@ class PrometheusPlugin(MonitorPlugin):
                 # 创建新的仪表
                 label_names = list(labels.keys()) if labels else []
                 metric = self._get_or_create_metric(name, MetricType.GAUGE, label_names)
+                if metric is None:
+                    return
                 if labels:
                     metric.labels(**labels).set(value)
                 else:
@@ -239,11 +247,15 @@ class PrometheusPlugin(MonitorPlugin):
                 # 创建新的直方图
                 label_names = list(labels.keys()) if labels else []
                 if buckets:
+                    if len(self._metrics) >= self._max_dynamic_metrics:
+                        return
                     metric = Histogram(
                         name, f"Auto-created histogram {name}", label_names, buckets=buckets
                     )
                 else:
                     metric = self._get_or_create_metric(name, MetricType.HISTOGRAM, label_names)
+                if metric is None:
+                    return
                 self._metrics[name] = metric
 
                 if labels:

src/services/health/monitor.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import os
+from collections import deque
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
@@ -118,10 +119,10 @@ class HealthMonitor:
 
     # === 其他配置 ===
     ALLOW_AUTO_RECOVER = os.getenv("HEALTH_AUTO_RECOVER_ENABLED", "true").lower() == "true"
-    CIRCUIT_HISTORY_LIMIT = int(os.getenv("HEALTH_CIRCUIT_HISTORY_LIMIT", "200"))
-
     # 进程级别状态缓存
-    _circuit_history: list[dict[str, Any]] = []
+    _circuit_history: deque[dict[str, Any]] = deque(
+        maxlen=int(os.getenv("HEALTH_CIRCUIT_HISTORY_LIMIT", "200"))
+    )
     _open_circuit_keys: int = 0
 
     # ==================== 数据访问辅助方法 ====================
@@ -865,14 +866,12 @@ class HealthMonitor:
     @classmethod
     def _push_circuit_event(cls, event: dict[str, Any]) -> None:
         cls._circuit_history.append(event)
-        if len(cls._circuit_history) > cls.CIRCUIT_HISTORY_LIMIT:
-            cls._circuit_history.pop(0)
 
     @classmethod
     def get_circuit_history(cls, limit: int = 50) -> list[dict[str, Any]]:
         if limit <= 0:
             return []
-        return cls._circuit_history[-limit:]
+        return list(cls._circuit_history)[-limit:]
 
     # ==================== 兼容旧方法 ====================
 

src/services/rate_limit/concurrency_manager.py

@@ -521,13 +521,12 @@ class ConcurrencyManager:
 
                 # 记录槽位占用时长分布
                 concurrency_slot_duration_seconds.labels(
-                    key_id=key_id[:8] if key_id else "unknown",  # 只记录前8位
                     exception=str(exception_occurred),
                 ).observe(slot_duration)
 
                 # 记录槽位释放计数
                 concurrency_slot_release_total.labels(
-                    key_id=key_id[:8] if key_id else "unknown", exception=str(exception_occurred)
+                    exception=str(exception_occurred),
                 ).inc()
 
                 # 告警：槽位占用时间过长（超过 60 秒）

src/services/task/execute/error_handler.py

@@ -55,7 +55,6 @@ class TaskErrorOperationsService:
         *,
         converted_error: Any,
         provider_type: str | None,
-        model_name: str | None,
         request_id: str | None,
         candidate_record_id: str,
         elapsed_ms: int,
@@ -141,7 +140,6 @@ class TaskErrorOperationsService:
 
                     antigravity_degradation_total.labels(
                         stage=stage_label,
-                        model=str(model_name or "unknown"),
                     ).inc()
                 except Exception:
                     pass
@@ -420,7 +418,6 @@ class TaskErrorOperationsService:
                 action = self.handle_thinking_signature_error(
                     converted_error=converted_error,
                     provider_type=str(getattr(provider, "provider_type", "") or "").lower(),
-                    model_name=str(global_model_id or ""),
                     request_id=request_id,
                     candidate_record_id=candidate_record_id,
                     elapsed_ms=elapsed_ms,