mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-07-28 08:44:41 +08:00
Feat/auto failover (#367)
* feat(db): add circuit breaker config table and provider proxy target APIs Add database support for auto-failover feature: - Add circuit_breaker_config table for storing failover thresholds - Add get/update_circuit_breaker_config methods in proxy DAO - Add reset_provider_health method for manual recovery - Add set_proxy_target and get_proxy_targets methods in providers DAO for managing multi-provider failover configuration * feat(proxy): implement circuit breaker and provider router for auto-failover Add core failover logic: - CircuitBreaker: Tracks provider health with three states: - Closed: Normal operation, requests pass through - Open: Circuit broken after consecutive failures, skip provider - HalfOpen: Testing recovery with limited requests - ProviderRouter: Routes requests across multiple providers with: - Health tracking and automatic failover - Configurable failure/success thresholds - Auto-disable proxy target after reaching failure threshold - Support for manual circuit breaker reset - Export new types in proxy module * feat(proxy): add failover Tauri commands and integrate with forwarder Expose failover functionality to frontend: - Add Tauri commands: get_proxy_targets, set_proxy_target, get_provider_health, reset_circuit_breaker, get/update_circuit_breaker_config, get_circuit_breaker_stats - Register all new commands in lib.rs invoke handler - Update forwarder with improved error handling and logging - Integrate ProviderRouter with proxy server startup - Add provider health tracking in request handlers * feat(frontend): add failover API layer and TanStack Query hooks Add frontend data layer for failover management: - Add failover.ts API: Tauri invoke wrappers for all failover commands - Add failover.ts query hooks: TanStack Query mutations and queries - useProxyTargets, useProviderHealth queries - useSetProxyTarget, useResetCircuitBreaker mutations - useCircuitBreakerConfig query and mutation - Update queries.ts with provider health query key - Update mutations.ts to invalidate health on provider changes - Add CircuitBreakerConfig and ProviderHealth types * feat(ui): add auto-failover configuration UI and provider health display Add comprehensive UI for failover management: Components: - ProviderHealthBadge: Display provider health status with color coding - CircuitBreakerConfigPanel: Configure failure/success thresholds, timeout duration, and error rate limits - AutoFailoverConfigPanel: Manage proxy targets with drag-and-drop priority ordering and individual enable/disable controls - ProxyPanel: Integrate failover tabs for unified proxy management Provider enhancements: - ProviderCard: Show health badge and proxy target indicator - ProviderActions: Add "Set as Proxy Target" action - EditProviderDialog: Add is_proxy_target toggle - ProviderList: Support proxy target filtering mode Other: - Update App.tsx routing for settings integration - Update useProviderActions hook with proxy target mutation - Fix ProviderList tests for updated component API * fix(usage): stabilize date range to prevent infinite re-renders * feat(backend): add tool version check command Add get_tool_versions command to check local and latest versions of Claude, Codex, and Gemini CLI tools: - Detect local installed versions via command line execution - Fetch latest versions from npm registry (Claude, Gemini) and GitHub releases API (Codex) - Return comprehensive version info including error details for uninstalled tools - Register command in Tauri invoke handler * style(ui): format accordion component code style Apply consistent code formatting to accordion component: - Convert double quotes to semicolons at line endings - Adjust indentation to 2-space standard - Align with project code style conventions * refactor(providers): update provider card styling to use theme tokens Replace hardcoded color classes with semantic design tokens: - Use bg-card, border-border, text-card-foreground instead of glass-card - Replace gray/white color literals with muted/foreground tokens - Change proxy target indicator color from purple to green - Improve hover states with border-border-active - Ensure consistent dark mode support via CSS variables * refactor(proxy): simplify auto-failover config panel structure Restructure AutoFailoverConfigPanel for better integration: - Remove internal Card wrapper and expansion toggle (now handled by parent) - Extract enabled state to props for external control - Simplify loading state display - Clean up redundant CardHeader/CardContent wrappers - ProxyPanel: reduce complexity by delegating to parent components * feat(settings): enhance settings page with accordion layout and tool versions Major settings page improvements: AboutSection: - Add local tool version detection (Claude, Codex, Gemini) - Display installed vs latest version comparison with visual indicators - Show update availability badges and environment check cards SettingsPage: - Reorganize advanced settings into collapsible accordion sections - Add proxy control panel with inline status toggle - Integrate auto-failover configuration with accordion UI - Add database and cost calculation config sections DirectorySettings & WindowSettings: - Minor styling adjustments for consistency settings.ts API: - Add getToolVersions() wrapper for new backend command * refactor(usage): restructure usage dashboard components Comprehensive usage statistics panel refactoring: UsageDashboard: - Reorganize layout with improved section headers - Add better loading states and empty state handling ModelStatsTable & ProviderStatsTable: - Minor styling updates for consistency ModelTestConfigPanel & PricingConfigPanel: - Simplify component structure - Remove redundant Card wrappers - Improve form field organization RequestLogTable: - Enhance table layout with better column sizing - Improve pagination controls UsageSummaryCards: - Update card styling with semantic tokens - Better responsive grid layout UsageTrendChart: - Refine chart container styling - Improve legend and tooltip display * chore(deps): add accordion and animation dependencies Package updates: - Add @radix-ui/react-accordion for collapsible sections - Add cmdk for command palette support - Add framer-motion for enhanced animations Tailwind config: - Add accordion-up/accordion-down animations - Update darkMode config to support both selector and class - Reorganize color and keyframe definitions for clarity * style(app): update header and app switcher styling App.tsx: - Replace glass-header with explicit bg-background/80 backdrop-blur - Update navigation button container to use bg-muted AppSwitcher: - Replace hardcoded gray colors with semantic muted/foreground tokens - Ensure consistent dark mode support via CSS variables - Add group class for better hover state transitions
This commit is contained in:
YoVinchen
@@ -0,0 +1,334 @@
|
||||
//! 熔断器模块
|
||||
//!
|
||||
//! 实现熔断器模式，用于防止向不健康的供应商发送请求
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
/// 熔断器状态
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum CircuitState {
|
||||
/// 关闭状态 - 正常工作
|
||||
Closed,
|
||||
/// 打开状态 - 熔断激活，拒绝请求
|
||||
Open,
|
||||
/// 半开状态 - 尝试恢复，允许部分请求通过
|
||||
HalfOpen,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for CircuitState {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
CircuitState::Closed => write!(f, "closed"),
|
||||
CircuitState::Open => write!(f, "open"),
|
||||
CircuitState::HalfOpen => write!(f, "half_open"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 熔断器配置
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct CircuitBreakerConfig {
|
||||
/// 失败阈值 - 连续失败多少次后打开熔断器
|
||||
pub failure_threshold: u32,
|
||||
/// 成功阈值 - 半开状态下成功多少次后关闭熔断器
|
||||
pub success_threshold: u32,
|
||||
/// 超时时间 - 熔断器打开后多久尝试半开（秒）
|
||||
pub timeout_seconds: u64,
|
||||
/// 错误率阈值 - 错误率超过此值时打开熔断器 (0.0-1.0)
|
||||
pub error_rate_threshold: f64,
|
||||
/// 最小请求数 - 计算错误率前的最小请求数
|
||||
pub min_requests: u32,
|
||||
}
|
||||
|
||||
impl Default for CircuitBreakerConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
failure_threshold: 5,
|
||||
success_threshold: 2,
|
||||
timeout_seconds: 60,
|
||||
error_rate_threshold: 0.5,
|
||||
min_requests: 10,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 熔断器实例
|
||||
pub struct CircuitBreaker {
|
||||
/// 当前状态
|
||||
state: Arc<RwLock<CircuitState>>,
|
||||
/// 连续失败计数
|
||||
consecutive_failures: Arc<AtomicU32>,
|
||||
/// 连续成功计数（半开状态）
|
||||
consecutive_successes: Arc<AtomicU32>,
|
||||
/// 总请求计数
|
||||
total_requests: Arc<AtomicU32>,
|
||||
/// 失败请求计数
|
||||
failed_requests: Arc<AtomicU32>,
|
||||
/// 上次打开时间
|
||||
last_opened_at: Arc<RwLock<Option<Instant>>>,
|
||||
/// 配置
|
||||
config: CircuitBreakerConfig,
|
||||
}
|
||||
|
||||
impl CircuitBreaker {
|
||||
/// 创建新的熔断器
|
||||
pub fn new(config: CircuitBreakerConfig) -> Self {
|
||||
Self {
|
||||
state: Arc::new(RwLock::new(CircuitState::Closed)),
|
||||
consecutive_failures: Arc::new(AtomicU32::new(0)),
|
||||
consecutive_successes: Arc::new(AtomicU32::new(0)),
|
||||
total_requests: Arc::new(AtomicU32::new(0)),
|
||||
failed_requests: Arc::new(AtomicU32::new(0)),
|
||||
last_opened_at: Arc::new(RwLock::new(None)),
|
||||
config,
|
||||
}
|
||||
}
|
||||
|
||||
/// 检查是否允许请求通过
|
||||
pub async fn allow_request(&self) -> bool {
|
||||
let state = *self.state.read().await;
|
||||
|
||||
match state {
|
||||
CircuitState::Closed => true,
|
||||
CircuitState::Open => {
|
||||
// 检查是否应该尝试半开
|
||||
if let Some(opened_at) = *self.last_opened_at.read().await {
|
||||
if opened_at.elapsed().as_secs() >= self.config.timeout_seconds {
|
||||
log::info!(
|
||||
"Circuit breaker transitioning from Open to HalfOpen (timeout reached)"
|
||||
);
|
||||
self.transition_to_half_open().await;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
CircuitState::HalfOpen => true,
|
||||
}
|
||||
}
|
||||
|
||||
/// 记录成功
|
||||
pub async fn record_success(&self) {
|
||||
let state = *self.state.read().await;
|
||||
|
||||
// 重置失败计数
|
||||
self.consecutive_failures.store(0, Ordering::SeqCst);
|
||||
self.total_requests.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
match state {
|
||||
CircuitState::HalfOpen => {
|
||||
let successes = self.consecutive_successes.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
log::debug!(
|
||||
"Circuit breaker HalfOpen: {} consecutive successes (threshold: {})",
|
||||
successes,
|
||||
self.config.success_threshold
|
||||
);
|
||||
|
||||
if successes >= self.config.success_threshold {
|
||||
log::info!("Circuit breaker transitioning from HalfOpen to Closed (success threshold reached)");
|
||||
self.transition_to_closed().await;
|
||||
}
|
||||
}
|
||||
CircuitState::Closed => {
|
||||
log::debug!("Circuit breaker Closed: request succeeded");
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// 记录失败
|
||||
pub async fn record_failure(&self) {
|
||||
let state = *self.state.read().await;
|
||||
|
||||
// 更新计数器
|
||||
let failures = self.consecutive_failures.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
self.total_requests.fetch_add(1, Ordering::SeqCst);
|
||||
self.failed_requests.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
// 重置成功计数
|
||||
self.consecutive_successes.store(0, Ordering::SeqCst);
|
||||
|
||||
log::debug!(
|
||||
"Circuit breaker {:?}: {} consecutive failures (threshold: {})",
|
||||
state,
|
||||
failures,
|
||||
self.config.failure_threshold
|
||||
);
|
||||
|
||||
// 检查是否应该打开熔断器
|
||||
match state {
|
||||
CircuitState::Closed | CircuitState::HalfOpen => {
|
||||
// 检查连续失败次数
|
||||
if failures >= self.config.failure_threshold {
|
||||
log::warn!(
|
||||
"Circuit breaker opening due to {} consecutive failures (threshold: {})",
|
||||
failures,
|
||||
self.config.failure_threshold
|
||||
);
|
||||
self.transition_to_open().await;
|
||||
} else {
|
||||
// 检查错误率
|
||||
let total = self.total_requests.load(Ordering::SeqCst);
|
||||
let failed = self.failed_requests.load(Ordering::SeqCst);
|
||||
|
||||
if total >= self.config.min_requests {
|
||||
let error_rate = failed as f64 / total as f64;
|
||||
log::debug!(
|
||||
"Circuit breaker error rate: {:.2}% ({}/{} requests)",
|
||||
error_rate * 100.0,
|
||||
failed,
|
||||
total
|
||||
);
|
||||
|
||||
if error_rate >= self.config.error_rate_threshold {
|
||||
log::warn!(
|
||||
"Circuit breaker opening due to high error rate: {:.2}% (threshold: {:.2}%)",
|
||||
error_rate * 100.0,
|
||||
self.config.error_rate_threshold * 100.0
|
||||
);
|
||||
self.transition_to_open().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// 获取当前状态
|
||||
pub async fn get_state(&self) -> CircuitState {
|
||||
*self.state.read().await
|
||||
}
|
||||
|
||||
/// 获取统计信息
|
||||
#[allow(dead_code)]
|
||||
pub async fn get_stats(&self) -> CircuitBreakerStats {
|
||||
CircuitBreakerStats {
|
||||
state: *self.state.read().await,
|
||||
consecutive_failures: self.consecutive_failures.load(Ordering::SeqCst),
|
||||
consecutive_successes: self.consecutive_successes.load(Ordering::SeqCst),
|
||||
total_requests: self.total_requests.load(Ordering::SeqCst),
|
||||
failed_requests: self.failed_requests.load(Ordering::SeqCst),
|
||||
}
|
||||
}
|
||||
|
||||
/// 重置熔断器（手动恢复）
|
||||
#[allow(dead_code)]
|
||||
pub async fn reset(&self) {
|
||||
log::info!("Circuit breaker manually reset to Closed state");
|
||||
self.transition_to_closed().await;
|
||||
}
|
||||
|
||||
/// 转换到打开状态
|
||||
async fn transition_to_open(&self) {
|
||||
*self.state.write().await = CircuitState::Open;
|
||||
*self.last_opened_at.write().await = Some(Instant::now());
|
||||
self.consecutive_failures.store(0, Ordering::SeqCst);
|
||||
self.consecutive_successes.store(0, Ordering::SeqCst);
|
||||
}
|
||||
|
||||
/// 转换到半开状态
|
||||
async fn transition_to_half_open(&self) {
|
||||
*self.state.write().await = CircuitState::HalfOpen;
|
||||
self.consecutive_successes.store(0, Ordering::SeqCst);
|
||||
}
|
||||
|
||||
/// 转换到关闭状态
|
||||
async fn transition_to_closed(&self) {
|
||||
*self.state.write().await = CircuitState::Closed;
|
||||
self.consecutive_failures.store(0, Ordering::SeqCst);
|
||||
self.consecutive_successes.store(0, Ordering::SeqCst);
|
||||
// 重置计数器
|
||||
self.total_requests.store(0, Ordering::SeqCst);
|
||||
self.failed_requests.store(0, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
|
||||
/// 熔断器统计信息
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct CircuitBreakerStats {
|
||||
pub state: CircuitState,
|
||||
pub consecutive_failures: u32,
|
||||
pub consecutive_successes: u32,
|
||||
pub total_requests: u32,
|
||||
pub failed_requests: u32,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_circuit_breaker_closed_to_open() {
|
||||
let config = CircuitBreakerConfig {
|
||||
failure_threshold: 3,
|
||||
..Default::default()
|
||||
};
|
||||
let breaker = CircuitBreaker::new(config);
|
||||
|
||||
// 初始状态应该是关闭
|
||||
assert_eq!(breaker.get_state().await, CircuitState::Closed);
|
||||
assert!(breaker.allow_request().await);
|
||||
|
||||
// 记录 3 次失败
|
||||
for _ in 0..3 {
|
||||
breaker.record_failure().await;
|
||||
}
|
||||
|
||||
// 应该转换到打开状态
|
||||
assert_eq!(breaker.get_state().await, CircuitState::Open);
|
||||
assert!(!breaker.allow_request().await);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_circuit_breaker_half_open_to_closed() {
|
||||
let config = CircuitBreakerConfig {
|
||||
failure_threshold: 2,
|
||||
success_threshold: 2,
|
||||
..Default::default()
|
||||
};
|
||||
let breaker = CircuitBreaker::new(config);
|
||||
|
||||
// 打开熔断器
|
||||
breaker.record_failure().await;
|
||||
breaker.record_failure().await;
|
||||
assert_eq!(breaker.get_state().await, CircuitState::Open);
|
||||
|
||||
// 手动转换到半开状态
|
||||
breaker.transition_to_half_open().await;
|
||||
assert_eq!(breaker.get_state().await, CircuitState::HalfOpen);
|
||||
|
||||
// 记录 2 次成功
|
||||
breaker.record_success().await;
|
||||
breaker.record_success().await;
|
||||
|
||||
// 应该转换到关闭状态
|
||||
assert_eq!(breaker.get_state().await, CircuitState::Closed);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_circuit_breaker_reset() {
|
||||
let config = CircuitBreakerConfig {
|
||||
failure_threshold: 2,
|
||||
..Default::default()
|
||||
};
|
||||
let breaker = CircuitBreaker::new(config);
|
||||
|
||||
// 打开熔断器
|
||||
breaker.record_failure().await;
|
||||
breaker.record_failure().await;
|
||||
assert_eq!(breaker.get_state().await, CircuitState::Open);
|
||||
|
||||
// 重置
|
||||
breaker.reset().await;
|
||||
assert_eq!(breaker.get_state().await, CircuitState::Closed);
|
||||
assert!(breaker.allow_request().await);
|
||||
}
|
||||
}
|
||||
@@ -4,8 +4,8 @@
|
||||
|
||||
use super::{
|
||||
error::*,
|
||||
provider_router::ProviderRouter as NewProviderRouter,
|
||||
providers::{get_adapter, ProviderAdapter},
|
||||
router::ProviderRouter,
|
||||
types::ProxyStatus,
|
||||
ProxyError,
|
||||
};
|
||||
@@ -18,9 +18,11 @@ use tokio::sync::RwLock;
|
||||
|
||||
pub struct RequestForwarder {
|
||||
client: Client,
|
||||
router: ProviderRouter,
|
||||
router: Arc<NewProviderRouter>,
|
||||
#[allow(dead_code)]
|
||||
max_retries: u8,
|
||||
status: Arc<RwLock<ProxyStatus>>,
|
||||
current_providers: Arc<RwLock<std::collections::HashMap<String, (String, String)>>>,
|
||||
}
|
||||
|
||||
impl RequestForwarder {
|
||||
@@ -29,6 +31,7 @@ impl RequestForwarder {
|
||||
timeout_secs: u64,
|
||||
max_retries: u8,
|
||||
status: Arc<RwLock<ProxyStatus>>,
|
||||
current_providers: Arc<RwLock<std::collections::HashMap<String, (String, String)>>>,
|
||||
) -> Self {
|
||||
let mut client_builder = Client::builder();
|
||||
if timeout_secs > 0 {
|
||||
@@ -41,13 +44,14 @@ impl RequestForwarder {
|
||||
|
||||
Self {
|
||||
client,
|
||||
router: ProviderRouter::new(db),
|
||||
router: Arc::new(NewProviderRouter::new(db)),
|
||||
max_retries,
|
||||
status,
|
||||
current_providers,
|
||||
}
|
||||
}
|
||||
|
||||
/// 转发请求（带重试和故障转移）
|
||||
/// 转发请求（带故障转移）
|
||||
pub async fn forward_with_retry(
|
||||
&self,
|
||||
app_type: &AppType,
|
||||
@@ -55,21 +59,39 @@ impl RequestForwarder {
|
||||
body: Value,
|
||||
headers: axum::http::HeaderMap,
|
||||
) -> Result<Response, ProxyError> {
|
||||
let mut failed_ids = Vec::new();
|
||||
let mut failover_happened = false;
|
||||
|
||||
// 获取适配器
|
||||
let adapter = get_adapter(app_type);
|
||||
let app_type_str = app_type.as_str();
|
||||
|
||||
for attempt in 0..self.max_retries {
|
||||
// 选择Provider
|
||||
let provider = self.router.select_provider(app_type, &failed_ids).await?;
|
||||
// 使用新的 ProviderRouter 选择所有可用供应商
|
||||
let providers = self
|
||||
.router
|
||||
.select_providers(app_type_str)
|
||||
.await
|
||||
.map_err(|e| ProxyError::DatabaseError(e.to_string()))?;
|
||||
|
||||
log::debug!(
|
||||
"尝试 {} - 使用Provider: {} ({})",
|
||||
if providers.is_empty() {
|
||||
return Err(ProxyError::NoAvailableProvider);
|
||||
}
|
||||
|
||||
log::info!(
|
||||
"[{}] 故障转移链: {} 个可用供应商",
|
||||
app_type_str,
|
||||
providers.len()
|
||||
);
|
||||
|
||||
let mut last_error = None;
|
||||
let mut failover_happened = false;
|
||||
|
||||
// 依次尝试每个供应商
|
||||
for (attempt, provider) in providers.iter().enumerate() {
|
||||
log::info!(
|
||||
"[{}] 尝试 {}/{} - 使用Provider: {} (sort_index: {})",
|
||||
app_type_str,
|
||||
attempt + 1,
|
||||
providers.len(),
|
||||
provider.name,
|
||||
provider.id
|
||||
provider.sort_index.unwrap_or(999999)
|
||||
);
|
||||
|
||||
// 更新状态中的当前Provider信息
|
||||
@@ -88,16 +110,29 @@ impl RequestForwarder {
|
||||
|
||||
// 转发请求
|
||||
match self
|
||||
.forward(&provider, endpoint, &body, &headers, adapter.as_ref())
|
||||
.forward(provider, endpoint, &body, &headers, adapter.as_ref())
|
||||
.await
|
||||
{
|
||||
Ok(response) => {
|
||||
let _latency = start.elapsed().as_millis() as u64;
|
||||
let latency = start.elapsed().as_millis() as u64;
|
||||
|
||||
// 成功：更新健康状态
|
||||
self.router
|
||||
.update_health(&provider, app_type, true, None)
|
||||
.await;
|
||||
// 成功：记录成功并更新熔断器
|
||||
if let Err(e) = self
|
||||
.router
|
||||
.record_result(&provider.id, app_type_str, true, None)
|
||||
.await
|
||||
{
|
||||
log::warn!("Failed to record success: {e}");
|
||||
}
|
||||
|
||||
// 更新当前应用类型使用的 provider
|
||||
{
|
||||
let mut current_providers = self.current_providers.write().await;
|
||||
current_providers.insert(
|
||||
app_type_str.to_string(),
|
||||
(provider.id.clone(), provider.name.clone()),
|
||||
);
|
||||
}
|
||||
|
||||
// 更新成功统计
|
||||
{
|
||||
@@ -106,6 +141,12 @@ impl RequestForwarder {
|
||||
status.last_error = None;
|
||||
if failover_happened {
|
||||
status.failover_count += 1;
|
||||
log::info!(
|
||||
"[{}] 故障转移成功！切换到 Provider: {} (耗时: {}ms)",
|
||||
app_type_str,
|
||||
provider.name,
|
||||
latency
|
||||
);
|
||||
}
|
||||
// 重新计算成功率
|
||||
if status.total_requests > 0 {
|
||||
@@ -115,23 +156,33 @@ impl RequestForwarder {
|
||||
}
|
||||
}
|
||||
|
||||
log::info!(
|
||||
"[{}] 请求成功 - Provider: {} - {}ms",
|
||||
app_type_str,
|
||||
provider.name,
|
||||
latency
|
||||
);
|
||||
|
||||
return Ok(response);
|
||||
}
|
||||
Err(e) => {
|
||||
let latency = start.elapsed().as_millis() as u64;
|
||||
|
||||
// 失败：分类错误
|
||||
// 失败：记录失败并更新熔断器
|
||||
if let Err(record_err) = self
|
||||
.router
|
||||
.record_result(&provider.id, app_type_str, false, Some(e.to_string()))
|
||||
.await
|
||||
{
|
||||
log::warn!("Failed to record failure: {record_err}");
|
||||
}
|
||||
|
||||
// 分类错误
|
||||
let category = self.categorize_proxy_error(&e);
|
||||
|
||||
match category {
|
||||
ErrorCategory::Retryable => {
|
||||
// 可重试：更新健康状态，添加到失败列表
|
||||
self.router
|
||||
.update_health(&provider, app_type, false, Some(e.to_string()))
|
||||
.await;
|
||||
failed_ids.push(provider.id.clone());
|
||||
|
||||
// 更新错误信息
|
||||
// 可重试：更新错误信息，继续尝试下一个供应商
|
||||
{
|
||||
let mut status = self.status.write().await;
|
||||
status.last_error =
|
||||
@@ -139,15 +190,19 @@ impl RequestForwarder {
|
||||
}
|
||||
|
||||
log::warn!(
|
||||
"请求失败（可重试）: Provider {} - {} - {}ms",
|
||||
"[{}] Provider {} 失败（可重试）: {} - {}ms",
|
||||
app_type_str,
|
||||
provider.name,
|
||||
e,
|
||||
latency
|
||||
);
|
||||
|
||||
last_error = Some(e);
|
||||
// 继续尝试下一个供应商
|
||||
continue;
|
||||
}
|
||||
ErrorCategory::NonRetryable | ErrorCategory::ClientAbort => {
|
||||
// 不可重试：更新失败统计并返回
|
||||
// 不可重试：直接返回错误
|
||||
{
|
||||
let mut status = self.status.write().await;
|
||||
status.failed_requests += 1;
|
||||
@@ -158,7 +213,12 @@ impl RequestForwarder {
|
||||
* 100.0;
|
||||
}
|
||||
}
|
||||
log::error!("请求失败（不可重试）: {e}");
|
||||
log::error!(
|
||||
"[{}] Provider {} 失败（不可重试）: {}",
|
||||
app_type_str,
|
||||
provider.name,
|
||||
e
|
||||
);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
@@ -166,18 +226,24 @@ impl RequestForwarder {
|
||||
}
|
||||
}
|
||||
|
||||
// 所有重试都失败
|
||||
// 所有供应商都失败了
|
||||
{
|
||||
let mut status = self.status.write().await;
|
||||
status.failed_requests += 1;
|
||||
status.last_error = Some("已达到最大重试次数".to_string());
|
||||
status.last_error = Some("所有供应商都失败".to_string());
|
||||
if status.total_requests > 0 {
|
||||
status.success_rate =
|
||||
(status.success_requests as f32 / status.total_requests as f32) * 100.0;
|
||||
}
|
||||
}
|
||||
|
||||
Err(ProxyError::MaxRetriesExceeded)
|
||||
log::error!(
|
||||
"[{}] 所有 {} 个供应商都失败了",
|
||||
app_type_str,
|
||||
providers.len()
|
||||
);
|
||||
|
||||
Err(last_error.unwrap_or(ProxyError::MaxRetriesExceeded))
|
||||
}
|
||||
|
||||
/// 转发单个请求（使用适配器）
|
||||
|
||||
@@ -322,6 +322,7 @@ pub async fn handle_messages(
|
||||
config.request_timeout,
|
||||
config.max_retries,
|
||||
state.status.clone(),
|
||||
state.current_providers.clone(),
|
||||
);
|
||||
|
||||
let response = forwarder
|
||||
@@ -641,6 +642,7 @@ pub async fn handle_gemini(
|
||||
config.request_timeout,
|
||||
config.max_retries,
|
||||
state.status.clone(),
|
||||
state.current_providers.clone(),
|
||||
);
|
||||
|
||||
// 提取完整的路径和查询参数
|
||||
@@ -806,6 +808,7 @@ pub async fn handle_responses(
|
||||
config.request_timeout,
|
||||
config.max_retries,
|
||||
state.status.clone(),
|
||||
state.current_providers.clone(),
|
||||
);
|
||||
|
||||
let response = forwarder
|
||||
@@ -985,6 +988,7 @@ pub async fn handle_chat_completions(
|
||||
config.request_timeout,
|
||||
config.max_retries,
|
||||
state.status.clone(),
|
||||
state.current_providers.clone(),
|
||||
);
|
||||
|
||||
let response = forwarder
|
||||
|
||||
@@ -2,10 +2,12 @@
|
||||
//!
|
||||
//! 提供本地HTTP代理服务，支持多Provider故障转移和请求透传
|
||||
|
||||
pub mod circuit_breaker;
|
||||
pub mod error;
|
||||
mod forwarder;
|
||||
mod handlers;
|
||||
mod health;
|
||||
pub mod provider_router;
|
||||
pub mod providers;
|
||||
pub mod response_handler;
|
||||
mod router;
|
||||
@@ -16,8 +18,14 @@ pub mod usage;
|
||||
|
||||
// 公开导出给外部使用（commands, services等模块需要）
|
||||
#[allow(unused_imports)]
|
||||
pub use circuit_breaker::{
|
||||
CircuitBreaker, CircuitBreakerConfig, CircuitBreakerStats, CircuitState,
|
||||
};
|
||||
#[allow(unused_imports)]
|
||||
pub use error::ProxyError;
|
||||
#[allow(unused_imports)]
|
||||
pub use provider_router::ProviderRouter;
|
||||
#[allow(unused_imports)]
|
||||
pub use response_handler::{NonStreamHandler, ResponseType, StreamHandler};
|
||||
#[allow(unused_imports)]
|
||||
pub use session::{ClientFormat, ProxySession};
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
//! 供应商路由器模块
|
||||
//!
|
||||
//! 负责选择和管理代理目标供应商，实现智能故障转移
|
||||
|
||||
use crate::database::Database;
|
||||
use crate::error::AppError;
|
||||
use crate::provider::Provider;
|
||||
use crate::proxy::circuit_breaker::CircuitBreaker;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
/// 供应商路由器
|
||||
pub struct ProviderRouter {
|
||||
/// 数据库连接
|
||||
db: Arc<Database>,
|
||||
/// 熔断器管理器 - key 格式: "app_type:provider_id"
|
||||
circuit_breakers: Arc<RwLock<HashMap<String, Arc<CircuitBreaker>>>>,
|
||||
}
|
||||
|
||||
impl ProviderRouter {
|
||||
/// 创建新的供应商路由器
|
||||
pub fn new(db: Arc<Database>) -> Self {
|
||||
Self {
|
||||
db,
|
||||
circuit_breakers: Arc::new(RwLock::new(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
/// 选择可用的供应商（支持故障转移）
|
||||
/// 返回按优先级排序的可用供应商列表
|
||||
pub async fn select_providers(&self, app_type: &str) -> Result<Vec<Provider>, AppError> {
|
||||
// 1. 获取所有启用代理的供应商
|
||||
let providers = self.db.get_proxy_targets(app_type).await?;
|
||||
|
||||
if providers.is_empty() {
|
||||
return Err(AppError::Config(
|
||||
"No proxy target providers configured".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
log::debug!(
|
||||
"Found {} proxy target providers for app_type: {}",
|
||||
providers.len(),
|
||||
app_type
|
||||
);
|
||||
|
||||
// 2. 按 sort_index 排序（已经在数据库查询中排序了）
|
||||
let sorted_providers: Vec<_> = providers.into_values().collect();
|
||||
|
||||
// 3. 过滤可用的供应商（检查熔断器状态）
|
||||
let mut available_providers = Vec::new();
|
||||
|
||||
for provider in sorted_providers {
|
||||
let circuit_key = format!("{}:{}", app_type, provider.id);
|
||||
let breaker = self.get_or_create_circuit_breaker(&circuit_key).await;
|
||||
|
||||
if breaker.allow_request().await {
|
||||
log::debug!(
|
||||
"Provider {} is available (circuit state: {:?})",
|
||||
provider.id,
|
||||
breaker.get_state().await
|
||||
);
|
||||
available_providers.push(provider);
|
||||
} else {
|
||||
log::warn!(
|
||||
"Provider {} is unavailable (circuit breaker open)",
|
||||
provider.id
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if available_providers.is_empty() {
|
||||
return Err(AppError::Config(
|
||||
"All proxy target providers are unavailable (circuit breakers open)".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
log::info!(
|
||||
"Selected {} available providers for failover chain",
|
||||
available_providers.len()
|
||||
);
|
||||
|
||||
Ok(available_providers)
|
||||
}
|
||||
|
||||
/// 记录供应商请求结果
|
||||
pub async fn record_result(
|
||||
&self,
|
||||
provider_id: &str,
|
||||
app_type: &str,
|
||||
success: bool,
|
||||
error_msg: Option<String>,
|
||||
) -> Result<(), AppError> {
|
||||
// 1. 更新熔断器状态
|
||||
let circuit_key = format!("{app_type}:{provider_id}");
|
||||
let breaker = self.get_or_create_circuit_breaker(&circuit_key).await;
|
||||
|
||||
if success {
|
||||
breaker.record_success().await;
|
||||
log::debug!("Provider {provider_id} request succeeded");
|
||||
} else {
|
||||
breaker.record_failure().await;
|
||||
log::warn!(
|
||||
"Provider {} request failed: {}",
|
||||
provider_id,
|
||||
error_msg.as_deref().unwrap_or("Unknown error")
|
||||
);
|
||||
}
|
||||
|
||||
// 2. 更新数据库健康状态
|
||||
self.db
|
||||
.update_provider_health(provider_id, app_type, success, error_msg.clone())
|
||||
.await?;
|
||||
|
||||
// 3. 如果连续失败达到熔断阈值，自动禁用代理目标
|
||||
if !success {
|
||||
let health = self.db.get_provider_health(provider_id, app_type).await?;
|
||||
|
||||
// 获取熔断器配置
|
||||
let config = self.db.get_circuit_breaker_config().await.ok();
|
||||
let failure_threshold = config.map(|c| c.failure_threshold).unwrap_or(5);
|
||||
|
||||
// 如果连续失败达到阈值，自动关闭该供应商的代理开关
|
||||
if health.consecutive_failures >= failure_threshold {
|
||||
log::warn!(
|
||||
"Provider {} has failed {} times (threshold: {}), auto-disabling proxy target",
|
||||
provider_id,
|
||||
health.consecutive_failures,
|
||||
failure_threshold
|
||||
);
|
||||
self.db
|
||||
.set_proxy_target(provider_id, app_type, false)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 重置熔断器（手动恢复）
|
||||
#[allow(dead_code)]
|
||||
pub async fn reset_circuit_breaker(&self, circuit_key: &str) {
|
||||
let breakers = self.circuit_breakers.read().await;
|
||||
if let Some(breaker) = breakers.get(circuit_key) {
|
||||
log::info!("Manually resetting circuit breaker for {circuit_key}");
|
||||
breaker.reset().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// 获取熔断器状态
|
||||
#[allow(dead_code)]
|
||||
pub async fn get_circuit_breaker_stats(
|
||||
&self,
|
||||
provider_id: &str,
|
||||
app_type: &str,
|
||||
) -> Option<crate::proxy::circuit_breaker::CircuitBreakerStats> {
|
||||
let circuit_key = format!("{app_type}:{provider_id}");
|
||||
let breakers = self.circuit_breakers.read().await;
|
||||
|
||||
if let Some(breaker) = breakers.get(&circuit_key) {
|
||||
Some(breaker.get_stats().await)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// 获取或创建熔断器
|
||||
async fn get_or_create_circuit_breaker(&self, key: &str) -> Arc<CircuitBreaker> {
|
||||
// 先尝试读锁获取
|
||||
{
|
||||
let breakers = self.circuit_breakers.read().await;
|
||||
if let Some(breaker) = breakers.get(key) {
|
||||
return breaker.clone();
|
||||
}
|
||||
}
|
||||
|
||||
// 如果不存在，获取写锁创建
|
||||
let mut breakers = self.circuit_breakers.write().await;
|
||||
|
||||
// 双重检查，防止竞争条件
|
||||
if let Some(breaker) = breakers.get(key) {
|
||||
return breaker.clone();
|
||||
}
|
||||
|
||||
// 从数据库加载配置
|
||||
let config = self
|
||||
.db
|
||||
.get_circuit_breaker_config()
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
|
||||
log::debug!("Creating new circuit breaker for {key} with config: {config:?}");
|
||||
|
||||
let breaker = Arc::new(CircuitBreaker::new(config));
|
||||
breakers.insert(key.to_string(), breaker.clone());
|
||||
|
||||
breaker
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::database::Database;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_provider_router_creation() {
|
||||
let db = Arc::new(Database::new_in_memory().unwrap());
|
||||
let router = ProviderRouter::new(db);
|
||||
|
||||
// 测试创建熔断器
|
||||
let breaker = router.get_or_create_circuit_breaker("claude:test").await;
|
||||
assert!(breaker.allow_request().await);
|
||||
}
|
||||
}
|
||||
@@ -57,6 +57,7 @@ impl ProviderRouter {
|
||||
}
|
||||
|
||||
/// 更新Provider健康状态（保留接口但不影响选择）
|
||||
#[allow(dead_code)]
|
||||
pub async fn update_health(
|
||||
&self,
|
||||
_provider: &Provider,
|
||||
|
||||
@@ -20,6 +20,8 @@ pub struct ProxyState {
|
||||
pub config: Arc<RwLock<ProxyConfig>>,
|
||||
pub status: Arc<RwLock<ProxyStatus>>,
|
||||
pub start_time: Arc<RwLock<Option<std::time::Instant>>>,
|
||||
/// 每个应用类型当前使用的 provider (app_type -> (provider_id, provider_name))
|
||||
pub current_providers: Arc<RwLock<std::collections::HashMap<String, (String, String)>>>,
|
||||
}
|
||||
|
||||
/// 代理HTTP服务器
|
||||
@@ -36,6 +38,7 @@ impl ProxyServer {
|
||||
config: Arc::new(RwLock::new(config.clone())),
|
||||
status: Arc::new(RwLock::new(ProxyStatus::default())),
|
||||
start_time: Arc::new(RwLock::new(None)),
|
||||
current_providers: Arc::new(RwLock::new(std::collections::HashMap::new())),
|
||||
};
|
||||
|
||||
Self {
|
||||
@@ -121,17 +124,16 @@ impl ProxyServer {
|
||||
status.uptime_seconds = start.elapsed().as_secs();
|
||||
}
|
||||
|
||||
// 获取所有活跃的代理目标
|
||||
if let Ok(targets) = self.state.db.get_all_proxy_targets() {
|
||||
status.active_targets = targets
|
||||
.into_iter()
|
||||
.map(|(app_type, name, id)| ActiveTarget {
|
||||
app_type,
|
||||
provider_name: name,
|
||||
provider_id: id,
|
||||
})
|
||||
.collect();
|
||||
}
|
||||
// 从 current_providers HashMap 获取每个应用类型当前正在使用的 provider
|
||||
let current_providers = self.state.current_providers.read().await;
|
||||
status.active_targets = current_providers
|
||||
.iter()
|
||||
.map(|(app_type, (provider_id, provider_name))| ActiveTarget {
|
||||
app_type: app_type.clone(),
|
||||
provider_id: provider_id.clone(),
|
||||
provider_name: provider_name.clone(),
|
||||
})
|
||||
.collect();
|
||||
|
||||
status
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user