feat: enhance volcengine quota sync with hard limits and smart buffering
This commit is contained in:
+204
-45
@@ -110,12 +110,21 @@ VOLC_MODEL_MAP = {
|
||||
# Populated by active_monitor_loop scanning config.yaml
|
||||
VOLC_MANAGED_PAIRS = set()
|
||||
|
||||
# Static Quotas (Fallback/Strict Limits)
|
||||
MODEL_QUOTAS = {
|
||||
# "deepseek-v3-2-251201:ba0383d3": 1600000,
|
||||
# "doubao-seed-1-8-251228:ba0383d3": 1800000,
|
||||
# "glm-4-7-251222:ba0383d3": 1800000,
|
||||
# "doubao-seed-1-8-251228:d57edef4": 1500000,
|
||||
# ==========================================
|
||||
# 🛑 Hard Daily Limits (Local Override / 本地每日硬限额)
|
||||
# ==========================================
|
||||
# Configured directly here to bypass API latency.
|
||||
# These limits are applied LOCALLY and update immediately.
|
||||
# 这里的限额是本地强制执行的，不受API延迟影响，用于防止超支。
|
||||
#
|
||||
# Format: "clean-model-name": limit_in_tokens
|
||||
# Or with hash: "clean-model-name:hash": limit_in_tokens
|
||||
HARD_DAILY_LIMITS = {
|
||||
# 示例 Examples:
|
||||
"doubao-seed-code-preview-251028": 5_000_000,
|
||||
"deepseek-v3-2-251201": 1_600_000,
|
||||
"doubao-seed-1-8-251228": 1_800_000,
|
||||
"glm-4-7-251222": 1_800_000,
|
||||
}
|
||||
|
||||
# ==========================================
|
||||
@@ -244,20 +253,20 @@ def get_usage_key(clean_model, api_base=None):
|
||||
|
||||
|
||||
def get_static_limit(clean_model, api_base):
|
||||
"""Check static configuration in MODEL_QUOTAS."""
|
||||
"""Check static configuration in HARD_DAILY_LIMITS."""
|
||||
current_hash = generate_api_base_hash(api_base)
|
||||
|
||||
# 1. Exact Match with Hash
|
||||
specific_key = f"{clean_model}:{current_hash}"
|
||||
if specific_key in MODEL_QUOTAS:
|
||||
return MODEL_QUOTAS[specific_key]
|
||||
if specific_key in HARD_DAILY_LIMITS:
|
||||
return HARD_DAILY_LIMITS[specific_key]
|
||||
|
||||
# 2. Exact Model Name
|
||||
if clean_model in MODEL_QUOTAS:
|
||||
return MODEL_QUOTAS[clean_model]
|
||||
if clean_model in HARD_DAILY_LIMITS:
|
||||
return HARD_DAILY_LIMITS[clean_model]
|
||||
|
||||
# 3. Fuzzy Match
|
||||
for quota_model, limit_val in MODEL_QUOTAS.items():
|
||||
for quota_model, limit_val in HARD_DAILY_LIMITS.items():
|
||||
if ":" not in quota_model and quota_model in clean_model:
|
||||
return limit_val
|
||||
return None
|
||||
@@ -265,53 +274,106 @@ def get_static_limit(clean_model, api_base):
|
||||
|
||||
def get_effective_limit(clean_model, api_base, current_usage=None):
|
||||
"""
|
||||
Calculate effective limit combining Static Quotas and Dynamic Redis Quotas.
|
||||
Dynamic Limit = Current Usage + Redis Remaining Balance
|
||||
Calculate effective limit using the Peak-Buffer strategy.
|
||||
Target Limit = (Peak Daily Remaining) - Buffer.
|
||||
"""
|
||||
if not redis_client:
|
||||
return get_static_limit(clean_model, api_base)
|
||||
|
||||
# 1. Get Static Limit
|
||||
# 1. Get Static Limit (Hard Config)
|
||||
static_limit = get_static_limit(clean_model, api_base)
|
||||
|
||||
# 2. Get Dynamic Remaining Balance
|
||||
# Key set by daemon loop: limit:dynamic:volcengine:remaining:{clean_model}
|
||||
dynamic_key = f"limit:dynamic:volcengine:remaining:{clean_model}"
|
||||
dynamic_remaining = redis_client.get(dynamic_key)
|
||||
# 2. Get Dynamic Stable Limit (Calculated by Daemon)
|
||||
# This limit is derived from: Peak_Daily_Balance - Buffer
|
||||
# It is stable and immune to API latency spikes.
|
||||
stable_limit_key = f"limit:dynamic:volcengine:stable_limit:{clean_model}"
|
||||
stable_limit_val = redis_client.get(stable_limit_key)
|
||||
|
||||
effective_limit = None
|
||||
|
||||
# If we have dynamic data
|
||||
if dynamic_remaining is not None:
|
||||
if stable_limit_val is not None:
|
||||
try:
|
||||
# If usage not provided, fetch it
|
||||
if current_usage is None:
|
||||
usage_key = get_usage_key(clean_model, api_base)
|
||||
current_usage = float(redis_client.get(usage_key) or 0)
|
||||
dynamic_limit = float(stable_limit_val)
|
||||
|
||||
# Dynamic Limit = The point where usage would consume all remaining balance
|
||||
# limit = usage + remaining
|
||||
dynamic_limit_val = float(current_usage) + float(dynamic_remaining)
|
||||
|
||||
# If static limit exists, take the stricter one (minimum)
|
||||
# UNLESS static is much higher? No, min implies stricter.
|
||||
# Apply Static Limit override if stricter
|
||||
if static_limit:
|
||||
effective_limit = min(float(static_limit), dynamic_limit_val)
|
||||
effective_limit = min(float(static_limit), dynamic_limit)
|
||||
else:
|
||||
effective_limit = dynamic_limit_val
|
||||
effective_limit = dynamic_limit
|
||||
|
||||
except (ValueError, TypeError):
|
||||
effective_limit = static_limit
|
||||
else:
|
||||
# Fallback: If daemon hasn't calculated stable limit yet, use old logic?
|
||||
# Or just use static. Safer to wait for daemon.
|
||||
effective_limit = static_limit
|
||||
|
||||
# Legacy fallback (Optional: Enable if you want immediate loose limits before first sync)
|
||||
# dynamic_remaining = redis_client.get(f"limit:dynamic:volcengine:remaining:{clean_model}")
|
||||
# if dynamic_remaining and current_usage:
|
||||
# effective_limit = float(current_usage) + float(dynamic_remaining)
|
||||
|
||||
return effective_limit
|
||||
|
||||
|
||||
# ==========================================
|
||||
# 👻 Handler Logic (DailyLimitHandler)
|
||||
# ==========================================
|
||||
# ... (eviction logic omitted for brevity) ...
|
||||
def soft_evict_from_router(target_model, target_api_base):
|
||||
"""
|
||||
Temporarily disable a model in the router by renaming it.
|
||||
"""
|
||||
if not LITELLM_AVAILABLE:
|
||||
return
|
||||
|
||||
try:
|
||||
from litellm.proxy.proxy_server import llm_router
|
||||
|
||||
if not llm_router:
|
||||
return
|
||||
|
||||
# Prepare target identifiers
|
||||
target_clean = clean_model_name(target_model)
|
||||
target_hash = generate_api_base_hash(target_api_base)
|
||||
|
||||
for i, model_entry in enumerate(llm_router.model_list):
|
||||
# Skip if already evicted/backed up
|
||||
if "_daily_limit_backup" in model_entry:
|
||||
continue
|
||||
|
||||
params = model_entry.get("litellm_params", {})
|
||||
m_model = params.get("model", "")
|
||||
m_base = params.get("api_base", "")
|
||||
|
||||
# Check Match
|
||||
m_clean = clean_model_name(m_model)
|
||||
m_hash = generate_api_base_hash(m_base)
|
||||
|
||||
# Match on Name AND Hash (Strict)
|
||||
if m_clean == target_clean and m_hash == target_hash:
|
||||
dprint(
|
||||
f"🚫 Soft-evicting {model_entry.get('model_name')} (Limit Reached)"
|
||||
)
|
||||
|
||||
# Create Backup
|
||||
model_entry["_daily_limit_backup"] = {
|
||||
"model_name": model_entry.get("model_name"),
|
||||
"api_base": m_base,
|
||||
}
|
||||
|
||||
# Disable by renaming (Router won't find it)
|
||||
model_entry["model_name"] = (
|
||||
f"{model_entry.get('model_name')}-limit-reached"
|
||||
)
|
||||
|
||||
# Optional: break after finding first match?
|
||||
# Or continue to evict all instances (e.g. load balancing same model)?
|
||||
# Let's evict ALL matches to be safe.
|
||||
# return
|
||||
|
||||
except Exception as e:
|
||||
log(f"❌ Soft evict failed: {e}", level="ERROR")
|
||||
|
||||
|
||||
last_restore_check = 0
|
||||
|
||||
@@ -471,7 +533,7 @@ class DailyLimitHandler(CustomLogger):
|
||||
# Display Real Balance (Remaining + Buffer)
|
||||
rem_int = int(float(rem_val))
|
||||
bal_est = rem_int + BUFFER
|
||||
rem_display = f", Bal: {bal_est:,}"
|
||||
rem_display = f" | Rem: {bal_est:,}"
|
||||
except Exception as e:
|
||||
print(
|
||||
f"DEBUG: Error calculating balance for {clean_model}: {e}"
|
||||
@@ -484,7 +546,7 @@ class DailyLimitHandler(CustomLogger):
|
||||
|
||||
key_display = key.split(":", 2)[-1]
|
||||
log(
|
||||
f"📊 [Usage] {key_display}: +{total_tokens} (Today: {current_total:,}{display_limit}{rem_display})",
|
||||
f"📊 [Usage] {key_display}: +{total_tokens} (Today: {current_total:,}{rem_display})",
|
||||
level="INFO",
|
||||
)
|
||||
|
||||
@@ -797,16 +859,22 @@ def active_monitor_loop():
|
||||
# < 100w -> 60min (3600s)
|
||||
# >= 100w -> 60min (3600s)
|
||||
|
||||
# Determine Base Sync Interval based on global risk
|
||||
# < 100k -> 5 min (Emergency but limit API calls due to latency)
|
||||
# < 300k -> 5 min (Safe Sailing)
|
||||
# < 500k -> 10 min (Speed Mode)
|
||||
# < 800k -> 30 min (Normal Mode)
|
||||
# >= 800k -> 60 min (High Tide Mode)
|
||||
if global_min_remaining < 100_000:
|
||||
base_interval = 120
|
||||
base_interval = 300 # 5 minutes (Increased from 2m)
|
||||
elif global_min_remaining < 300_000:
|
||||
base_interval = 300
|
||||
base_interval = 300 # 5 minutes
|
||||
elif global_min_remaining < 500_000:
|
||||
base_interval = 600
|
||||
base_interval = 600 # 10 minutes
|
||||
elif global_min_remaining < 800_000:
|
||||
base_interval = 1800
|
||||
base_interval = 1800 # 30 minutes
|
||||
else:
|
||||
base_interval = 3600
|
||||
base_interval = 3600 # 60 minutes
|
||||
|
||||
if is_idle:
|
||||
current_sync_interval = min(base_interval * 4, 3600)
|
||||
@@ -815,18 +883,60 @@ def active_monitor_loop():
|
||||
else:
|
||||
current_sync_interval = base_interval
|
||||
|
||||
# =====================================================
|
||||
# 🚀 Per-Model Burst Detection (精细化突发流量检测)
|
||||
# =====================================================
|
||||
burst_triggered = False
|
||||
burst_reason = ""
|
||||
|
||||
for model in INTERESTED_MODELS:
|
||||
# 1. Get Current Usage
|
||||
curr_usage_key = f"limit:dynamic:volcengine:usage:{model}"
|
||||
curr_usage_val = redis_client.get(curr_usage_key)
|
||||
curr_usage = float(curr_usage_val) if curr_usage_val else 0.0
|
||||
|
||||
# 2. Get Last Synced Usage (Snapshot)
|
||||
last_usage = last_usage_map.get(model, 0.0)
|
||||
|
||||
# 3. Calculate Delta since last sync
|
||||
model_delta = curr_usage - last_usage
|
||||
|
||||
# 4. Get Current Balance for Threshold Calculation
|
||||
rem_key = f"limit:dynamic:volcengine:remaining:{model}"
|
||||
rem_val = redis_client.get(rem_key)
|
||||
current_balance = float(rem_val) if rem_val else 0.0
|
||||
|
||||
# 5. Determine Dynamic Threshold based on Balance
|
||||
# Rich models get loose limits, Poor models get strict limits.
|
||||
if current_balance > 1000000:
|
||||
model_threshold = 500000
|
||||
elif current_balance > 500000:
|
||||
model_threshold = 100000
|
||||
elif current_balance > 100000:
|
||||
model_threshold = 50000
|
||||
else:
|
||||
model_threshold = 10000 # Strict for poor models
|
||||
|
||||
# 6. Check Trigger
|
||||
if model_delta > model_threshold:
|
||||
burst_triggered = True
|
||||
burst_reason = f"{model} (+{model_delta:,} > {model_threshold:,})"
|
||||
break # One burst is enough to trigger sync
|
||||
|
||||
should_sync = (
|
||||
(delta > THRESHOLD)
|
||||
burst_triggered
|
||||
or (time_diff > current_sync_interval)
|
||||
or (last_sync_time == 0)
|
||||
)
|
||||
|
||||
if should_sync:
|
||||
trigger_reason = (
|
||||
"Burst Traffic" if delta > THRESHOLD else "Timer Expired"
|
||||
f"Burst: {burst_reason}" if burst_triggered else "Timer Expired"
|
||||
)
|
||||
# For Timer Expired, we don't have a specific delta to show, or we can calculate global delta if needed.
|
||||
# But for now, let's just show the reason.
|
||||
print(
|
||||
f"{'💤' if is_idle else '⚡'} Sync Triggered ({trigger_reason}) / 同步触发... Delta: {delta}, Interval: {current_sync_interval}s",
|
||||
f"{'💤' if is_idle else '⚡'} Sync Triggered ({trigger_reason}) / 同步触发... Interval: {current_sync_interval}s",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
@@ -867,6 +977,46 @@ def active_monitor_loop():
|
||||
# Format: limit:dynamic:volcengine:remaining:{model_name}
|
||||
dynamic_key = f"limit:dynamic:volcengine:remaining:{target}"
|
||||
|
||||
# ==========================================
|
||||
# 🏔️ Peak Tracking & Stable Limit
|
||||
# ==========================================
|
||||
# Track the highest remaining balance seen today.
|
||||
# Limit = Peak * 0.85 (15% Safety Buffer)
|
||||
# This prevents "Expanding Limit" bugs caused by API latency.
|
||||
peak_key = f"volc:peak_balance:{today_str}:{target}"
|
||||
current_peak = float(redis_client.get(peak_key) or 0)
|
||||
|
||||
if balance > current_peak:
|
||||
if DEBUG_MODE:
|
||||
print(
|
||||
f"DEBUG: New Peak for {target}: {int(balance):,}",
|
||||
flush=True,
|
||||
)
|
||||
redis_client.setex(
|
||||
peak_key, 172800, int(balance)
|
||||
) # 48h retention
|
||||
current_peak = balance
|
||||
|
||||
# Calculate Stable Limit
|
||||
# Logic: Limit = Peak - Buffer
|
||||
# Buffer = Max(Peak * 15%, 250,000)
|
||||
# This ensures at least 250k safety margin, or 15% for large packages.
|
||||
safety_margin = max(current_peak * 0.15, 250000)
|
||||
stable_limit = int(max(0, current_peak - safety_margin))
|
||||
|
||||
stable_limit_key = (
|
||||
f"limit:dynamic:volcengine:stable_limit:{target}"
|
||||
)
|
||||
|
||||
# Store stable limit
|
||||
redis_client.setex(stable_limit_key, 86400, int(stable_limit))
|
||||
|
||||
if DEBUG_MODE:
|
||||
print(
|
||||
f"DEBUG: {target} Peak={int(current_peak):,} | Buffer={int(safety_margin):,} | Limit={int(stable_limit):,}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
if DEBUG_MODE:
|
||||
print(
|
||||
f"DEBUG: Checking {target} (Configured / 已配置) (Bal: {int(balance):,}, Rem: {int(remaining):,})",
|
||||
@@ -918,7 +1068,16 @@ def active_monitor_loop():
|
||||
else:
|
||||
global_min_remaining = temp_min_remaining
|
||||
|
||||
last_usage_map["global"] = current_usage_global
|
||||
# Update snapshots for next iteration delta calculation
|
||||
current_usage_global_snapshot = 0
|
||||
for m in INTERESTED_MODELS:
|
||||
uk = f"limit:dynamic:volcengine:usage:{m}"
|
||||
uv = redis_client.get(uk)
|
||||
u_val = float(uv) if uv else 0.0
|
||||
last_usage_map[m] = u_val
|
||||
current_usage_global_snapshot += u_val
|
||||
|
||||
last_usage_map["global"] = current_usage_global_snapshot
|
||||
last_sync_time = now
|
||||
|
||||
if global_min_remaining > 800000:
|
||||
|
||||
Reference in New Issue
Block a user