i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
- 移除 token/latency Prometheus 指标的 model 标签，避免 provider x model 笛卡尔积
- HealthMonitor 滑动窗口从 DB JSON 迁移至进程内存，减少写放大
- ModelCostService 三层缓存增加 500 条上限，超限时清空
- StickyPriority 粘性缓存和健康状态字典增加容量淘汰
- AffinityManager 请求锁字典增加 500 条上限，淘汰空闲锁
- 配额刷新/探测查询使用 defer/load_only 避免加载大 JSON 列
- Alembic 迁移清理 DB 中遗留的 request_results_window 数据
- 同步更新测试适配 batch_get_cooldowns 返回值和批量删除异步化
This commit is contained in:
fawney19
2026-03-10 10:40:50 +08:00
parent c9f0685b40
commit 7c580e843f
20 changed files with 241 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/plugins/load_balancer/sticky_priority.py
View File

@@ -78,10 +78,12 @@ class StickyPriorityStrategy(LoadBalancerStrategy):
"total_failures": 0,
}
)
self._max_provider_health_entries: int = 200 # 上限，防止无界增长
# 当前粘性提供商缓存 {cache_key: provider_id}
# cache_key 可以是 api_key_id 或者其他标识
self._sticky_providers: dict[str, str] = {}
self._max_sticky_entries: int = 2000 # 上限，防止无界增长
# 统计信息
self._stats = {
@@ -260,6 +262,12 @@ class StickyPriorityStrategy(LoadBalancerStrategy):
# 没有缓存或缓存失效，选择权重最大的
sticky_candidate = max(candidates, key=lambda c: c.weight)
# 超出上限时清理不在当前候选列表中的旧条目
if len(self._sticky_providers) >= self._max_sticky_entries:
active_ids = {str(c.provider.id) for c in candidates}
stale = [k for k, v in self._sticky_providers.items() if v not in active_ids]
for k in stale[: len(stale) // 2 or len(stale)]:
del self._sticky_providers[k]
self._sticky_providers[cache_key] = str(sticky_candidate.provider.id)
logger.info(f"Set new sticky provider {sticky_candidate.provider.name}")
@@ -358,6 +366,17 @@ class StickyPriorityStrategy(LoadBalancerStrategy):
error: 错误信息（如果失败）
"""
provider_id = str(provider.id)
# 超出上限时淘汰健康且请求量最少的条目
if (
provider_id not in self._provider_health
and len(self._provider_health) >= self._max_provider_health_entries
):
healthy_keys = [
k for k, v in self._provider_health.items() if v["is_healthy"] and k != provider_id
]
if healthy_keys:
victim = min(healthy_keys, key=lambda k: self._provider_health[k]["total_requests"])
del self._provider_health[victim]
health_info = self._provider_health[provider_id]
health_info["total_requests"] += 1

6
src/plugins/monitor/base.py
View File

@@ -209,7 +209,7 @@ class MonitorPlugin(BasePlugin):
def record_token_usage(
self,
provider: str,
model: str,
model: str, # noqa: ARG002 - 保留签名兼容性，不再用于 Prometheus 标签
input_tokens: int,
output_tokens: int,
cost: float | None = None,
@@ -219,12 +219,12 @@ class MonitorPlugin(BasePlugin):
Args:
provider: 提供商名称
model: 模型名称
model: 模型名称（保留签名兼容性，不再作为 Prometheus 标签）
input_tokens: 输入token数
output_tokens: 输出token数
cost: 费用
"""
labels = {"provider": provider, "model": model}
labels = {"provider": provider}
import asyncio

14
src/plugins/monitor/prometheus.py
View File

@@ -74,21 +74,19 @@ class PrometheusPlugin(MonitorPlugin):
http_label_names,
)
# Token使用指标
# Token使用指标（仅按 provider 维度聚合，避免 provider x model 笛卡尔积导致基数爆炸）
self._metrics["tokens_input_total"] = Counter(
"tokens_input_total", "Total input tokens", ["provider", "model"]
"tokens_input_total", "Total input tokens", ["provider"]
)
self._metrics["tokens_output_total"] = Counter(
"tokens_output_total", "Total output tokens", ["provider", "model"]
"tokens_output_total", "Total output tokens", ["provider"]
)
self._metrics["tokens_total"] = Counter(
"tokens_total", "Total tokens", ["provider", "model"]
)
self._metrics["tokens_total"] = Counter("tokens_total", "Total tokens", ["provider"])
self._metrics["usage_cost_total"] = Counter(
"usage_cost_total", "Total usage cost in USD", ["provider", "model"]
"usage_cost_total", "Total usage cost in USD", ["provider"]
)
# 系统指标
@@ -112,7 +110,7 @@ class PrometheusPlugin(MonitorPlugin):
self._metrics["provider_latency_seconds"] = Histogram(
"provider_latency_seconds",
"Provider response latency in seconds",
["provider", "model"],
["provider"],
buckets=(0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30),
)

66
src/services/health/monitor.py
View File

@@ -37,13 +37,12 @@ class CircuitState:
HALF_OPEN = "half_open" # 半开（验证恢复）
# 默认健康度数据结构
# 默认健康度数据结构（不含 request_results_window，窗口数据仅存进程内存）
def _default_health_data() -> dict[str, Any]:
return {
"health_score": 1.0,
"consecutive_failures": 0,
"last_failure_at": None,
"request_results_window": [],
}
@@ -125,6 +124,12 @@ class HealthMonitor:
)
_open_circuit_keys: int = 0
# === 滑动窗口进程内存缓存 ===
# Key: (key_id, api_format), Value: list of {"ts": float, "ok": bool}
# 不再持久化到数据库，进程重启后自然重建
_window_cache: dict[tuple[str, str], list[dict[str, Any]]] = {}
_WINDOW_CACHE_MAX_ENTRIES = int(os.getenv("HEALTH_WINDOW_CACHE_MAX_ENTRIES", "10000"))
# ==================== 数据访问辅助方法 ====================
@classmethod
@@ -137,11 +142,43 @@ class HealthMonitor:
@classmethod
def _set_health_data(cls, key: ProviderAPIKey, api_format: str, data: dict[str, Any]) -> None:
"""设置指定格式的健康度数据"""
"""设置指定格式的健康度数据（写入 DB 前剥离窗口数据）"""
health_by_format = dict(key.health_by_format or {})
health_by_format[api_format] = data
db_data = {k: v for k, v in data.items() if k != "request_results_window"}
health_by_format[api_format] = db_data
key.health_by_format = health_by_format # type: ignore[assignment]
# ==================== 滑动窗口进程内存缓存方法 ====================
@classmethod
def _get_window(cls, key_id: str, api_format: str) -> list[dict[str, Any]]:
"""从进程内存缓存获取滑动窗口"""
return cls._window_cache.get((key_id, api_format), [])
@classmethod
def _set_window(cls, key_id: str, api_format: str, window: list[dict[str, Any]]) -> None:
"""设置滑动窗口到进程内存缓存（带容量淘汰）"""
cache_key = (key_id, api_format)
if (
cache_key not in cls._window_cache
and len(cls._window_cache) >= cls._WINDOW_CACHE_MAX_ENTRIES
):
try:
oldest_key = next(iter(cls._window_cache))
del cls._window_cache[oldest_key]
except StopIteration:
pass
cls._window_cache[cache_key] = window
@classmethod
def _clear_window(cls, key_id: str, api_format: str | None = None) -> None:
"""清理滑动窗口缓存"""
if api_format:
cls._window_cache.pop((key_id, api_format), None)
else:
for k in [k for k in cls._window_cache if k[0] == key_id]:
del cls._window_cache[k]
@classmethod
def _get_circuit_data(cls, key: ProviderAPIKey, api_format: str) -> dict[str, Any]:
"""获取指定格式的熔断器数据，不存在则返回默认值"""
@@ -208,14 +245,15 @@ class HealthMonitor:
health_data = cls._get_health_data(key, effective_api_format)
circuit_data = cls._get_circuit_data(key, effective_api_format)
# 1. 更新滑动窗口
window = health_data.get("request_results_window") or []
# 1. 更新滑动窗口（进程内存缓存，不持久化到 DB）
window = cls._get_window(key.id, effective_api_format)
window = list(window) # 避免原地修改
window.append({"ts": now_ts, "ok": True})
cutoff_ts = now_ts - cls.WINDOW_SECONDS
window = [r for r in window if r["ts"] > cutoff_ts]
if len(window) > cls.WINDOW_SIZE:
window = window[-cls.WINDOW_SIZE :]
health_data["request_results_window"] = window
cls._set_window(key.id, effective_api_format, window)
# 2. 更新健康度（用于展示）
current_score = float(health_data.get("health_score") or 0)
@@ -333,14 +371,15 @@ class HealthMonitor:
health_data = cls._get_health_data(key, effective_api_format)
circuit_data = cls._get_circuit_data(key, effective_api_format)
# 1. 更新滑动窗口
window = health_data.get("request_results_window") or []
# 1. 更新滑动窗口（进程内存缓存，不持久化到 DB）
window = cls._get_window(key.id, effective_api_format)
window = list(window) # 避免原地修改
window.append({"ts": now_ts, "ok": False})
cutoff_ts = now_ts - cls.WINDOW_SECONDS
window = [r for r in window if r["ts"] > cutoff_ts]
if len(window) > cls.WINDOW_SIZE:
window = window[-cls.WINDOW_SIZE :]
health_data["request_results_window"] = window
cls._set_window(key.id, effective_api_format, window)
# 2. 更新健康度（用于展示）
current_score = float(health_data.get("health_score") or 1)
@@ -654,7 +693,7 @@ class HealthMonitor:
# 查询单个格式
health_data = cls._get_health_data(key, api_format)
circuit_data = cls._get_circuit_data(key, api_format)
window = health_data.get("request_results_window") or []
window = cls._get_window(key_id, api_format)
valid_window = [r for r in window if r["ts"] > now_ts - cls.WINDOW_SECONDS]
result["api_format"] = api_format
@@ -678,7 +717,7 @@ class HealthMonitor:
for fmt in key.api_formats or []:
health_data = health_by_format.get(fmt, _default_health_data())
circuit_data = circuit_by_format.get(fmt, _default_circuit_data())
window = health_data.get("request_results_window") or []
window = cls._get_window(key_id, fmt)
valid_window = [r for r in window if r["ts"] > now_ts - cls.WINDOW_SECONDS]
formats_health[fmt] = {
@@ -755,11 +794,13 @@ class HealthMonitor:
# 重置单个格式
cls._set_health_data(key, api_format, _default_health_data())
cls._set_circuit_data(key, api_format, _default_circuit_data())
cls._clear_window(key_id, api_format)
logger.info(f"[RESET] 重置 Key 健康度: {key_id}/{api_format}")
else:
# 重置所有格式
key.health_by_format = {} # type: ignore[assignment]
key.circuit_breaker_by_format = {} # type: ignore[assignment]
cls._clear_window(key_id)
logger.info(f"[RESET] 重置 Key 所有格式健康度: {key_id}")
db.flush()
@@ -788,6 +829,7 @@ class HealthMonitor:
# 重置所有格式的健康度
key.health_by_format = {} # type: ignore[assignment]
key.circuit_breaker_by_format = {} # type: ignore[assignment]
cls._clear_window(key_id)
logger.info(f"[OK] 手动启用 Key: {key_id}")
db.flush()

13
src/services/model/cost.py
View File

@@ -55,6 +55,7 @@ class ModelCostService:
_price_cache: dict[str, dict[str, float]] = {}
_cache_price_cache: dict[str, dict[str, float]] = {}
_tiered_pricing_cache: dict[str, dict | None] = {}
_MAX_CACHE_SIZE: int = 500 # 每层缓存上限，防止无界增长
def __init__(self, db: Session):
self.db = db
@@ -200,6 +201,7 @@ class ModelCostService:
"source": "global",
}
self._evict_if_full(self._tiered_pricing_cache)
self._tiered_pricing_cache[cache_key] = result
return result
@@ -258,6 +260,7 @@ class ModelCostService:
"source": "global",
}
self._evict_if_full(self._tiered_pricing_cache)
self._tiered_pricing_cache[cache_key] = result
return result.get("pricing") if result else None
@@ -359,6 +362,7 @@ class ModelCostService:
f"未找到模型价格配置: {provider_name}/{model}，请在 GlobalModel 中配置价格"
)
self._evict_if_full(self._price_cache)
self._price_cache[cache_key] = {"input": input_price, "output": output_price}
return input_price, output_price
@@ -440,6 +444,7 @@ class ModelCostService:
model,
)
self._evict_if_full(self._price_cache)
self._price_cache[cache_key] = {"input": input_price, "output": output_price}
return input_price, output_price
@@ -521,6 +526,7 @@ class ModelCostService:
if cache_read_price is None:
cache_read_price = input_price * 0.1
self._evict_if_full(self._cache_price_cache)
self._cache_price_cache[cache_key] = {
"creation": cache_creation_price,
"read": cache_read_price,
@@ -687,6 +693,7 @@ class ModelCostService:
if cache_read_price is None:
cache_read_price = input_price * 0.1
self._evict_if_full(self._cache_price_cache)
self._cache_price_cache[cache_key] = {
"creation": cache_creation_price,
"read": cache_read_price,
@@ -873,6 +880,12 @@ class ModelCostService:
cls._cache_price_cache.clear()
cls._tiered_pricing_cache.clear()
@classmethod
def _evict_if_full(cls, cache: dict) -> None:
"""缓存超出上限时清空，防止无界增长。"""
if len(cache) >= cls._MAX_CACHE_SIZE:
cache.clear()
# ------------------------------------------------------------------
# 内部辅助
# ------------------------------------------------------------------

15
src/services/provider_keys/key_quota_service.py
View File

@@ -6,7 +6,7 @@ import asyncio
from collections.abc import Awaitable, Callable
from typing import Any
from sqlalchemy.orm import Session
from sqlalchemy.orm import Session, defer
from src.core.exceptions import InvalidRequestException, NotFoundException
from src.core.logger import logger
@@ -101,8 +101,17 @@ async def refresh_provider_quota_for_provider(
deduped.append(value)
selected_key_ids = deduped
keys_query = db.query(ProviderAPIKey).filter(
ProviderAPIKey.provider_id == provider_id,
keys_query = (
db.query(ProviderAPIKey)
.options(
defer(ProviderAPIKey.health_by_format),
defer(ProviderAPIKey.circuit_breaker_by_format),
defer(ProviderAPIKey.adjustment_history),
defer(ProviderAPIKey.utilization_samples),
)
.filter(
ProviderAPIKey.provider_id == provider_id,
)
)
if selected_key_ids is None:
keys_query = keys_query.filter(ProviderAPIKey.is_active.is_(True))

10
src/services/provider_keys/pool_quota_probe_scheduler.py
View File

@@ -15,6 +15,8 @@ from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from sqlalchemy.orm import load_only
from src.clients.redis_client import get_redis_client
from src.core.logger import logger
from src.core.provider_types import ProviderType, normalize_provider_type
@@ -275,6 +277,14 @@ class PoolQuotaProbeScheduler:
if eligible_ids:
all_keys = (
db.query(ProviderAPIKey)
.options(
load_only(
ProviderAPIKey.id,
ProviderAPIKey.provider_id,
ProviderAPIKey.last_used_at,
ProviderAPIKey.upstream_metadata,
)
)
.filter(
ProviderAPIKey.provider_id.in_(eligible_ids),
ProviderAPIKey.is_active == True, # noqa: E712

6
src/services/scheduling/affinity_manager.py
View File

@@ -108,6 +108,7 @@ class CacheAffinityManager:
# 请求级别锁，避免同一用户+端点同时更新造成抖动
self._request_locks: dict[str, asyncio.Lock] = {}
self._request_locks_max_size: int = 500 # 锁字典上限，防止无界增长
# 统计信息
self._stats = {
@@ -209,6 +210,11 @@ class CacheAffinityManager:
async def _acquire_request_lock(self, cache_key: str) -> None:
lock = self._request_locks.get(cache_key)
if lock is None:
# 超出上限时淘汰未被持有的锁，防止无界增长
if len(self._request_locks) >= self._request_locks_max_size:
free_keys = [k for k, lk in self._request_locks.items() if not lk.locked()]
for k in free_keys[: len(free_keys) // 2 or 1]: # 清理一半空闲锁
del self._request_locks[k]
lock = asyncio.Lock()
self._request_locks[cache_key] = lock
await lock.acquire()