feat(proxy): 节点状态简化、连接事件记录、可靠性指标与批量删除

- 移除 UNHEALTHY 中间状态,节点状态简化为 ONLINE/OFFLINE
- 新增 proxy_node_events 表记录 tunnel 连接/断开/错误事件
- 新增 failed_requests/dns_failures/stream_errors 可靠性指标(增量累加)
- tunnel 重连改为固定 1s 延迟,移除指数退避逻辑
- resolver/service 改为以 TunnelManager 内存状态判断节点可用性,避免 DB 竞态
- 修正 Claude cache_control 字段格式,使用 ttl 字段控制缓存时长
- 移除前端手动勾选 capability 的 UI,改为从价格配置自动推断
- 新增全局模型批量删除 API,替换前端并行单个删除
This commit is contained in:
fawney19
2026-02-28 13:52:32 +08:00
parent ecb16d345a
commit 54530faf03
25 changed files with 586 additions and 225 deletions

View File

@@ -202,19 +202,21 @@ pub struct Config {
#[arg(long, env = "AETHER_PROXY_LOG_JSON", default_value_t = false)]
pub log_json: bool,
/// WebSocket reconnect base delay in milliseconds
/// Deprecated: reconnect now uses a fixed 1s delay. Kept for config compatibility.
#[arg(
long,
env = "AETHER_PROXY_TUNNEL_RECONNECT_BASE_MS",
default_value_t = 500
default_value_t = 500,
hide = true
)]
pub tunnel_reconnect_base_ms: u64,
/// WebSocket reconnect max delay in milliseconds
/// Deprecated: reconnect now uses a fixed 1s delay. Kept for config compatibility.
#[arg(
long,
env = "AETHER_PROXY_TUNNEL_RECONNECT_MAX_MS",
default_value_t = 30000
default_value_t = 30000,
hide = true
)]
pub tunnel_reconnect_max_ms: u64,
@@ -288,13 +290,6 @@ impl Config {
if self.aether_retry_max_attempts == 0 {
anyhow::bail!("aether_retry_max_attempts must be >= 1");
}
if self.tunnel_reconnect_base_ms > self.tunnel_reconnect_max_ms {
anyhow::bail!(
"tunnel_reconnect_base_ms ({}) must be <= tunnel_reconnect_max_ms ({})",
self.tunnel_reconnect_base_ms,
self.tunnel_reconnect_max_ms
);
}
if self.upstream_connect_timeout_secs == 0 {
anyhow::bail!("upstream_connect_timeout_secs must be > 0");
}

View File

@@ -1,6 +1,5 @@
//! WebSocket tunnel client: connect, authenticate, and run the tunnel.
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -226,20 +225,6 @@ pub fn build_tls_config() -> rustls::ClientConfig {
.with_no_client_auth()
}
/// Calculate next reconnect delay with exponential backoff + jitter.
pub fn next_reconnect_delay(state: &Arc<AppState>, reconnect_attempts: &AtomicU32) -> Duration {
let attempt = reconnect_attempts.fetch_add(1, Ordering::Relaxed);
let base_ms = state.config.tunnel_reconnect_base_ms;
let max_ms = state.config.tunnel_reconnect_max_ms;
let delay_ms = base_ms.saturating_mul(1u64 << attempt.min(10)).min(max_ms);
let jitter = (delay_ms / 4).max(1);
let jitter_ms = rand_u64() % jitter;
Duration::from_millis(delay_ms + jitter_ms)
}
fn build_tunnel_url(server: &ServerContext) -> String {
let base = server.aether_url.trim_end_matches('/');
let ws_base = if base.starts_with("https://") {
@@ -251,20 +236,3 @@ fn build_tunnel_url(server: &ServerContext) -> String {
};
format!("{}/api/internal/proxy-tunnel", ws_base)
}
/// Simple pseudo-random u64 (no external crate needed).
fn rand_u64() -> u64 {
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let seed = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as u64;
let cnt = COUNTER.fetch_add(1, Ordering::Relaxed);
let mut x = seed ^ cnt;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
x
}

View File

@@ -5,7 +5,6 @@ pub mod protocol;
pub mod stream_handler;
pub mod writer;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -14,10 +13,9 @@ use tracing::{error, info};
use crate::state::{AppState, ServerContext};
/// Minimum connection duration (seconds) to consider a session "stable".
/// If a connection lasts shorter than this, the backoff counter is NOT reset,
/// preventing rapid reconnect loops on persistently bad networks.
const MIN_STABLE_DURATION: Duration = Duration::from_secs(30);
/// Fixed reconnect delay -- short enough for fast recovery, long enough to
/// avoid CPU spin when the network is completely down.
const RECONNECT_DELAY: Duration = Duration::from_secs(1);
/// Run the tunnel mode main loop (connect, dispatch, reconnect).
///
@@ -31,42 +29,17 @@ pub async fn run(
) {
info!(server = %server.server_label, conn = conn_idx, "starting tunnel");
// Per-connection reconnect counter (avoids N connections interfering
// with each other's backoff via the shared ServerContext field).
let reconnect_attempts = AtomicU32::new(0);
loop {
let connect_start = tokio::time::Instant::now();
match client::connect_and_run(state, server, conn_idx, &mut shutdown).await {
Ok(client::TunnelOutcome::Shutdown) => {
info!(server = %server.server_label, conn = conn_idx, "tunnel shut down gracefully");
return;
}
Ok(client::TunnelOutcome::Disconnected) => {
let duration = connect_start.elapsed();
if duration >= MIN_STABLE_DURATION {
// Stable session -- reset backoff for quick reconnect
reconnect_attempts.store(0, Ordering::Release);
info!(
server = %server.server_label,
conn = conn_idx,
duration_secs = duration.as_secs(),
"tunnel disconnected after stable session"
);
} else {
// Short-lived session -- keep backoff increasing
info!(
server = %server.server_label,
conn = conn_idx,
duration_secs = duration.as_secs(),
"tunnel disconnected quickly, increasing backoff"
);
}
info!(server = %server.server_label, conn = conn_idx, "tunnel disconnected, reconnecting");
}
Err(e) => {
// Connection failed -- keep backoff increasing
error!(server = %server.server_label, conn = conn_idx, error = %e, "tunnel connection lost");
error!(server = %server.server_label, conn = conn_idx, error = %e, "tunnel connection error, reconnecting");
}
}
@@ -75,11 +48,8 @@ pub async fn run(
return;
}
let delay = client::next_reconnect_delay(state, &reconnect_attempts);
info!(server = %server.server_label, conn = conn_idx, delay_ms = delay.as_millis(), "reconnecting tunnel");
tokio::select! {
_ = tokio::time::sleep(delay) => {}
_ = tokio::time::sleep(RECONNECT_DELAY) => {}
_ = shutdown.changed() => {
info!(server = %server.server_label, conn = conn_idx, "shutdown requested during reconnect wait");
return;

View File

@@ -0,0 +1,114 @@
"""proxy_node_metrics_and_events
Revision ID: 48afe197cc15
Revises: b2c3d4e5f6a7
Create Date: 2026-02-28 04:33:11.201185+00:00
"""
import sqlalchemy as sa
from sqlalchemy import inspect
from alembic import op
# revision identifiers, used by Alembic.
revision = "48afe197cc15"
down_revision = "b2c3d4e5f6a7"
branch_labels = None
depends_on = None
def _column_exists(table_name: str, column_name: str) -> bool:
bind = op.get_bind()
insp = inspect(bind)
columns = [c["name"] for c in insp.get_columns(table_name)]
return column_name in columns
def _table_exists(table_name: str) -> bool:
bind = op.get_bind()
insp = inspect(bind)
return table_name in insp.get_table_names()
def upgrade() -> None:
# proxy_nodes: 新增错误指标字段
if not _column_exists("proxy_nodes", "failed_requests"):
op.add_column(
"proxy_nodes",
sa.Column(
"failed_requests",
sa.BigInteger(),
nullable=False,
server_default="0",
comment="累计失败请求数",
),
)
if not _column_exists("proxy_nodes", "dns_failures"):
op.add_column(
"proxy_nodes",
sa.Column(
"dns_failures",
sa.BigInteger(),
nullable=False,
server_default="0",
comment="累计 DNS 失败数",
),
)
if not _column_exists("proxy_nodes", "stream_errors"):
op.add_column(
"proxy_nodes",
sa.Column(
"stream_errors",
sa.BigInteger(),
nullable=False,
server_default="0",
comment="累计流错误数",
),
)
# proxy_node_events: 连接事件表
if not _table_exists("proxy_node_events"):
op.create_table(
"proxy_node_events",
sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column("node_id", sa.String(length=36), nullable=False),
sa.Column(
"event_type",
sa.String(length=20),
nullable=False,
comment="事件类型: connected, disconnected, error",
),
sa.Column(
"detail",
sa.String(length=500),
nullable=True,
comment="事件详情(如断开原因)",
),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(["node_id"], ["proxy_nodes.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(
"idx_proxy_node_events_node_created",
"proxy_node_events",
["node_id", "created_at"],
)
op.create_index(
op.f("ix_proxy_node_events_node_id"),
"proxy_node_events",
["node_id"],
)
def downgrade() -> None:
if _table_exists("proxy_node_events"):
op.drop_index(op.f("ix_proxy_node_events_node_id"), table_name="proxy_node_events")
op.drop_index("idx_proxy_node_events_node_created", table_name="proxy_node_events")
op.drop_table("proxy_node_events")
if _column_exists("proxy_nodes", "stream_errors"):
op.drop_column("proxy_nodes", "stream_errors")
if _column_exists("proxy_nodes", "dns_failures"):
op.drop_column("proxy_nodes", "dns_failures")
if _column_exists("proxy_nodes", "failed_requests"):
op.drop_column("proxy_nodes", "failed_requests")

View File

@@ -68,6 +68,16 @@ export async function deleteGlobalModel(
await client.delete(`/api/admin/models/global/${id}`, { params: { force } })
}
/**
* 批量删除 GlobalModel
*/
export async function batchDeleteGlobalModels(
ids: string[]
): Promise<{ success_count: number; failed: Array<{ id: string; error: string }> }> {
const response = await client.post('/api/admin/models/global/batch-delete', { ids })
return response.data
}
/**
* 批量为 GlobalModel 添加关联提供商
*/

View File

@@ -13,7 +13,7 @@ export interface ProxyNode {
ip: string
port: number
region: string | null
status: 'online' | 'unhealthy' | 'offline'
status: 'online' | 'offline'
is_manual: boolean
tunnel_mode: boolean
tunnel_connected: boolean
@@ -34,10 +34,20 @@ export interface ProxyNode {
active_connections: number
total_requests: number
avg_latency_ms: number | null
failed_requests: number
dns_failures: number
stream_errors: number
created_at: string
updated_at: string
}
export interface ProxyNodeEvent {
id: number
event_type: 'connected' | 'disconnected' | 'error'
detail: string | null
created_at: string
}
export interface ProxyNodeListResponse {
items: ProxyNode[]
total: number
@@ -103,4 +113,9 @@ export const proxyNodesApi = {
const response = await apiClient.post<ProxyNodeTestResult>('/api/admin/proxy-nodes/test-url', data)
return response.data
},
async listNodeEvents(nodeId: string, limit = 50): Promise<{ items: ProxyNodeEvent[] }> {
const response = await apiClient.get<{ items: ProxyNodeEvent[] }>(`/api/admin/proxy-nodes/${nodeId}/events`, { params: { limit } })
return response.data
},
}

View File

@@ -179,31 +179,6 @@
</div>
</section>
<!-- Key 能力配置 -->
<section
v-if="availableCapabilities.length > 0"
class="space-y-2"
>
<h4 class="font-medium text-sm">
模型偏好
</h4>
<div class="flex flex-wrap gap-2">
<label
v-for="cap in availableCapabilities"
:key="cap.name"
class="flex items-center gap-2 px-2.5 py-1 rounded-md border bg-muted/30 cursor-pointer text-sm"
>
<input
type="checkbox"
:checked="form.supported_capabilities?.includes(cap.name)"
class="rounded"
@change="toggleCapability(cap.name)"
>
<span>{{ cap.display_name }}</span>
</label>
</div>
</section>
<!-- 价格配置 -->
<section class="space-y-3">
<h4 class="font-medium text-sm">
@@ -212,7 +187,7 @@
<TieredPricingEditor
ref="tieredPricingEditorRef"
v-model="tieredPricing"
:show-cache1h="form.supported_capabilities?.includes('cache_1h')"
:show-cache1h="true"
/>
<div class="flex items-center gap-3 pt-2 border-t">
<Label class="text-xs whitespace-nowrap">按次计费</Label>
@@ -353,7 +328,7 @@
</template>
<script setup lang="ts">
import { ref, computed, watch, onMounted } from 'vue'
import { ref, computed, watch } from 'vue'
import {
Loader2, Layers, SquarePen,
Search, ChevronRight, Plus, Trash2
@@ -378,7 +353,6 @@ import {
type GlobalModelUpdate,
} from '@/api/global-models'
import type { TieredPricingConfig } from '@/api/endpoints/types'
import { getAllCapabilities, type CapabilityDefinition } from '@/api/endpoints'
const props = defineProps<{
open: boolean
@@ -665,34 +639,6 @@ function fillVideoResolutionPricePreset(preset: 'common' | 'sora' | 'veo') {
}))
}
// Key 能力选项
const availableCapabilities = ref<CapabilityDefinition[]>([])
// 加载可用能力列表
async function loadCapabilities() {
try {
availableCapabilities.value = await getAllCapabilities()
} catch (err) {
log.error('Failed to load capabilities:', err)
}
}
// 切换能力
function toggleCapability(capName: string) {
if (!form.value.supported_capabilities) {
form.value.supported_capabilities = []
}
const index = form.value.supported_capabilities.indexOf(capName)
if (index >= 0) {
form.value.supported_capabilities.splice(index, 1)
} else {
form.value.supported_capabilities.push(capName)
}
}
onMounted(() => {
loadCapabilities()
})
// 加载模型列表
async function loadModels() {
@@ -829,6 +775,19 @@ async function handleSubmit() {
// Apply billing (video) pricing into config before cleaning/submitting.
applyVideoPricingToConfig()
// Auto-infer supported_capabilities from tiered pricing config
const caps = new Set(form.value.supported_capabilities || [])
const has1hPricing = finalTieredPricing?.tiers?.some(
(t: Record<string, unknown>) => Array.isArray(t.cache_ttl_pricing)
&& (t.cache_ttl_pricing as Array<Record<string, unknown>>).some(c => c.ttl_minutes === 60)
)
if (has1hPricing) {
caps.add('cache_1h')
} else {
caps.delete('cache_1h')
}
form.value.supported_capabilities = caps.size > 0 ? [...caps] : []
// 清理空的 config
const cleanConfig = form.value.config && Object.keys(form.value.config).length > 0
? form.value.config

View File

@@ -182,7 +182,7 @@
v-if="getFirst1hCachePrice(model.default_tiered_pricing) !== '-'"
class="flex items-center gap-3 p-3 rounded-lg border bg-muted/20"
>
<Label class="text-xs text-muted-foreground whitespace-nowrap">1h 缓存创建</Label>
<Label class="text-xs text-muted-foreground whitespace-nowrap">1h 缓存</Label>
<span class="text-sm font-mono">{{ getFirst1hCachePrice(model.default_tiered_pricing) }}</span>
</div>
<!-- 按次计费 -->

View File

@@ -112,7 +112,7 @@
v-if="showCache1h"
class="space-y-1"
>
<Label class="text-xs text-muted-foreground">1h 缓存创建</Label>
<Label class="text-xs text-muted-foreground">1h 缓存</Label>
<Input
:model-value="getCache1hDisplay(index)"
type="number"
@@ -209,26 +209,6 @@ watch(
{ immediate: true }
)
// 监听 showCache1h 变化
watch(
() => props.showCache1h,
(newValue, oldValue) => {
if (oldValue === true && newValue === false) {
// 取消勾选时,清除本地的 1h 缓存数据和手动设置标记
localTiers.value.forEach((tier, i) => {
tier.cache_ttl_pricing = undefined
if (cacheManuallySet[i]) {
cacheManuallySet[i].cache1h = false
}
})
syncToParent()
} else if (oldValue === false && newValue === true) {
// 勾选时,同步自动计算的价格到父组件
syncToParent()
}
}
)
// 验证错误
const validationError = computed(() => {
if (localTiers.value.length === 0) {

View File

@@ -245,18 +245,8 @@ const tieredPricingEditorRef = ref<InstanceType<typeof TieredPricingEditor> | nu
const isEditing = computed(() => !!props.editingModel)
// 计算是否显示 1h 缓存输入框
const showCache1h = computed(() => {
if (isEditing.value) {
// 编辑模式:检查当前配置是否有 1h 缓存配置(从 tiered_pricing 或 effective_tiered_pricing 中检测)
const pricing = props.editingModel?.tiered_pricing || props.editingModel?.effective_tiered_pricing
return pricing?.tiers?.some(t => t.cache_ttl_pricing?.some(c => c.ttl_minutes === 60)) ?? false
} else {
// 添加模式:从选中的全局模型中读取 supported_capabilities
const selectedModel = availableGlobalModels.value.find(m => m.id === form.value.global_model_id)
return selectedModel?.supported_capabilities?.includes('cache_1h') ?? false
}
})
// 1h 缓存定价始终显示
const showCache1h = true
// 表单状态
const submitting = ref(false)

View File

@@ -656,6 +656,7 @@ import {
getGlobalModel,
updateGlobalModel,
deleteGlobalModel,
batchDeleteGlobalModels,
batchAssignToProviders,
getGlobalModelProviders,
type GlobalModelResponse,
@@ -1276,15 +1277,13 @@ async function confirmBatchDeleteModels() {
submittingBatchManage.value = true
try {
const ids = Array.from(selectedBatchManageModelIds.value)
const results = await Promise.allSettled(ids.map(id => deleteGlobalModel(id)))
const successCount = results.filter(r => r.status === 'fulfilled').length
const failCount = results.filter(r => r.status === 'rejected').length
const result = await batchDeleteGlobalModels(ids)
if (successCount > 0) {
success(`成功删除 ${successCount} 个模型`)
if (result.success_count > 0) {
success(`成功删除 ${result.success_count} 个模型`)
}
if (failCount > 0) {
showError(`${failCount} 个模型删除失败`, '部分失败')
if (result.failed.length > 0) {
showError(`${result.failed.length} 个模型删除失败`, '部分失败')
}
// 清除选中的已删除模型

View File

@@ -48,9 +48,6 @@
<SelectItem value="online">
在线
</SelectItem>
<SelectItem value="unhealthy">
异常
</SelectItem>
<SelectItem value="offline">
离线
</SelectItem>
@@ -86,9 +83,6 @@
<SelectItem value="online">
在线
</SelectItem>
<SelectItem value="unhealthy">
异常
</SelectItem>
<SelectItem value="offline">
离线
</SelectItem>
@@ -237,6 +231,16 @@
>
<Settings class="h-4 w-4" />
</Button>
<Button
v-if="!node.is_manual"
variant="ghost"
size="icon"
class="h-8 w-8"
title="连接事件"
@click="handleViewEvents(node)"
>
<History class="h-4 w-4" />
</Button>
<Button
variant="ghost"
size="icon"
@@ -548,6 +552,77 @@
</Button>
</template>
</Dialog>
<!-- 连接事件对话框 -->
<Dialog
:open="showEventsDialog"
title="连接事件"
:description="eventsNode ? `${eventsNode.name} 的连接历史` : ''"
size="lg"
@update:open="(v: boolean) => { if (!v) { showEventsDialog = false; eventsNode = null; nodeEvents = [] } }"
>
<div class="space-y-3">
<!-- 可靠性指标摘要 -->
<div
v-if="eventsNode"
class="grid grid-cols-3 gap-3 text-sm"
>
<div class="bg-muted/40 rounded-lg px-3 py-2 text-center">
<span class="block text-foreground/60 text-xs">失败请求</span>
<span class="tabular-nums font-medium">{{ formatNumber(eventsNode.failed_requests || 0) }}</span>
</div>
<div class="bg-muted/40 rounded-lg px-3 py-2 text-center">
<span class="block text-foreground/60 text-xs">DNS 失败</span>
<span class="tabular-nums font-medium">{{ formatNumber(eventsNode.dns_failures || 0) }}</span>
</div>
<div class="bg-muted/40 rounded-lg px-3 py-2 text-center">
<span class="block text-foreground/60 text-xs">流错误</span>
<span class="tabular-nums font-medium">{{ formatNumber(eventsNode.stream_errors || 0) }}</span>
</div>
</div>
<!-- 事件列表 -->
<div
v-if="loadingEvents"
class="py-8 text-center text-muted-foreground text-sm"
>
加载中...
</div>
<div
v-else-if="nodeEvents.length === 0"
class="py-8 text-center text-muted-foreground text-sm"
>
暂无连接事件记录
</div>
<div
v-else
class="max-h-80 overflow-y-auto space-y-1.5"
>
<div
v-for="event in nodeEvents"
:key="event.id"
class="flex items-center gap-2 px-3 py-2 rounded-lg bg-muted/30 text-sm"
>
<Badge
:variant="eventTypeVariant(event.event_type)"
class="text-[10px] px-1.5 py-0 shrink-0"
>
{{ eventTypeLabel(event.event_type) }}
</Badge>
<span class="text-muted-foreground truncate flex-1">{{ event.detail || '-' }}</span>
<span class="text-xs text-muted-foreground/70 tabular-nums shrink-0">{{ formatTime(event.created_at) }}</span>
</div>
</div>
</div>
<template #footer>
<Button
variant="outline"
@click="showEventsDialog = false; eventsNode = null; nodeEvents = []"
>
关闭
</Button>
</template>
</Dialog>
</div>
</template>
@@ -556,7 +631,7 @@ import { ref, computed, onMounted, watch } from 'vue'
import { useProxyNodesStore } from '@/stores/proxy-nodes'
import { useToast } from '@/composables/useToast'
import { useConfirm } from '@/composables/useConfirm'
import { proxyNodesApi, type ProxyNode, type ProxyNodeRemoteConfig } from '@/api/proxy-nodes'
import { proxyNodesApi, type ProxyNode, type ProxyNodeRemoteConfig, type ProxyNodeEvent } from '@/api/proxy-nodes'
import {
Card,
@@ -580,7 +655,7 @@ import {
Dialog,
} from '@/components/ui'
import { Search, Trash2, Plus, SquarePen, Activity, Loader2, Settings } from 'lucide-vue-next'
import { Search, Trash2, Plus, SquarePen, Activity, Loader2, Settings, History } from 'lucide-vue-next'
import { parseApiError } from '@/utils/errorParser'
import { formatRegion } from '@/utils/region'
import HardwareTooltip from './components/HardwareTooltip.vue'
@@ -616,6 +691,12 @@ const configForm = ref({
heartbeat_interval: '30',
})
// 连接事件对话框
const showEventsDialog = ref(false)
const eventsNode = ref<ProxyNode | null>(null)
const nodeEvents = ref<ProxyNodeEvent[]>([])
const loadingEvents = ref(false)
// 测试连通性
const testingNodes = ref(new Set<string>())
const testingUrl = ref(false)
@@ -834,10 +915,41 @@ async function handleTest(node: ProxyNode) {
}
}
async function handleViewEvents(node: ProxyNode) {
eventsNode.value = node
showEventsDialog.value = true
loadingEvents.value = true
try {
const res = await proxyNodesApi.listNodeEvents(node.id, 50)
nodeEvents.value = res.items
} catch (err: unknown) {
toastError(parseApiError(err, '加载事件失败'))
} finally {
loadingEvents.value = false
}
}
function eventTypeLabel(type: string) {
switch (type) {
case 'connected': return '连接'
case 'disconnected': return '断开'
case 'error': return '错误'
default: return type
}
}
function eventTypeVariant(type: string) {
switch (type) {
case 'connected': return 'success' as const
case 'disconnected': return 'destructive' as const
case 'error': return 'destructive' as const
default: return 'secondary' as const
}
}
function statusVariant(status: string) {
switch (status) {
case 'online': return 'success' as const
case 'unhealthy': return 'secondary' as const
case 'offline': return 'destructive' as const
default: return 'secondary' as const
}
@@ -846,7 +958,6 @@ function statusVariant(status: string) {
function statusLabel(status: string) {
switch (status) {
case 'online': return '在线'
case 'unhealthy': return '异常'
case 'offline': return '离线'
default: return status
}

View File

@@ -192,7 +192,7 @@
v-if="getFirst1hCachePrice(model.default_tiered_pricing) !== '-'"
class="flex items-center gap-3 p-3 rounded-lg border bg-muted/20"
>
<Label class="text-xs text-muted-foreground whitespace-nowrap">1h 缓存创建</Label>
<Label class="text-xs text-muted-foreground whitespace-nowrap">1h 缓存</Label>
<span class="text-sm font-mono">{{ getFirst1hCachePrice(model.default_tiered_pricing) }}</span>
</div>
<!-- 按次计费 -->

View File

@@ -9,7 +9,7 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from fastapi import APIRouter, Depends, Query, Request, Response
from fastapi import APIRouter, Body, Depends, Query, Request, Response
from sqlalchemy.orm import Session
from src.api.base.admin_adapter import AdminApiAdapter
@@ -181,6 +181,21 @@ async def delete_global_model(
return Response(status_code=204)
@router.post("/batch-delete")
async def batch_delete_global_models(
request: Request,
ids: list[str] = Body(..., embed=True, max_length=100),
db: Session = Depends(get_db),
) -> dict:
"""
批量删除 GlobalModel
顺序删除多个 GlobalModel每个独立提交避免并行删除导致的锁竞争。
"""
adapter = AdminBatchDeleteGlobalModelsAdapter(ids=ids)
return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
@router.post(
"/{global_model_id}/assign-to-providers", response_model=BatchAssignToProvidersResponse
)
@@ -517,6 +532,50 @@ class AdminDeleteGlobalModelAdapter(AdminApiAdapter):
return None
@dataclass
class AdminBatchDeleteGlobalModelsAdapter(AdminApiAdapter):
"""批量删除多个 GlobalModel顺序执行每个删除独立提交"""
ids: list[str]
async def handle(self, context: ApiRequestContext) -> Any: # type: ignore[override]
from src.core.exceptions import NotFoundException
from src.models.database import GlobalModel
success_count = 0
failed: list[dict] = []
deleted_names: list[tuple[str, str]] = [] # (name, id)
for gm_id in self.ids:
try:
gm = context.db.query(GlobalModel).filter(GlobalModel.id == gm_id).first()
if gm:
name = gm.name
mid = gm.id
GlobalModelService.delete_global_model(context.db, gm_id)
deleted_names.append((name, mid))
success_count += 1
else:
failed.append({"id": gm_id, "error": "not found"})
except NotFoundException:
failed.append({"id": gm_id, "error": "not found"})
except Exception as e:
context.db.rollback()
failed.append({"id": gm_id, "error": str(e)})
# 批量失效缓存
if deleted_names:
from src.services.cache.invalidation import get_cache_invalidation_service
cache_service = get_cache_invalidation_service()
for name, mid in deleted_names:
await cache_service.on_global_model_changed(name, mid)
logger.info("批量删除 GlobalModel: success={}, failed={}", success_count, len(failed))
return {"success_count": success_count, "failed": failed}
@dataclass
class AdminBatchAssignToProvidersAdapter(AdminApiAdapter):
"""批量为 Provider 添加 GlobalModel 实现"""

View File

@@ -221,6 +221,17 @@ async def update_proxy_node_config(
return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
@router.get("/{node_id}/events")
async def list_proxy_node_events(
node_id: str,
request: Request,
limit: int = Query(50, ge=1, le=200),
db: Session = Depends(get_db),
) -> Any:
adapter = AdminListProxyNodeEventsAdapter(node_id=node_id, limit=limit)
return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
# ---------------------------------------------------------------------------
# 辅助函数
# ---------------------------------------------------------------------------
@@ -517,3 +528,34 @@ class AdminTestProxyUrlAdapter(AdminApiAdapter):
username=req.username,
password=req.password,
)
@dataclass
class AdminListProxyNodeEventsAdapter(AdminApiAdapter):
"""查询代理节点连接事件(连接/断开/错误历史)"""
name: str = "admin_list_proxy_node_events"
node_id: str = ""
limit: int = 50
async def handle(self, context: ApiRequestContext) -> Any:
from src.models.database import ProxyNodeEvent
events = (
context.db.query(ProxyNodeEvent)
.filter(ProxyNodeEvent.node_id == self.node_id)
.order_by(ProxyNodeEvent.created_at.desc())
.limit(self.limit)
.all()
)
return {
"items": [
{
"id": e.id,
"event_type": e.event_type,
"detail": e.detail,
"created_at": e.created_at,
}
for e in events
],
}

View File

@@ -119,6 +119,7 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
# 启动服务端 ping 任务,防止中间代理因空闲超时关闭连接
ping_task = asyncio.create_task(_ping_loop(conn))
disconnect_reason: str | None = None
try:
oversized_count = 0
while True:
@@ -126,6 +127,7 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
data = await asyncio.wait_for(ws.receive_bytes(), timeout=_IDLE_TIMEOUT)
except asyncio.TimeoutError:
logger.warning("tunnel idle timeout for node_id={}", node_id)
disconnect_reason = "idle timeout"
await ws.close(code=4004, reason="idle timeout")
break
if len(data) > _MAX_FRAME_SIZE:
@@ -133,6 +135,7 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
logger.warning("tunnel frame too large from {}: {} bytes", node_id, len(data))
if oversized_count >= 5:
logger.warning("too many oversized frames from {}, closing", node_id)
disconnect_reason = "too many oversized frames"
await ws.close(code=4003, reason="too many oversized frames")
break
continue
@@ -146,14 +149,16 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
await manager.handle_incoming_frame(conn, frame)
except WebSocketDisconnect:
disconnect_reason = "WebSocket disconnected"
logger.info("tunnel WebSocket disconnected: node_id={}", node_id)
except Exception as e:
disconnect_reason = f"error: {e}"
logger.error("tunnel WebSocket error for node_id={}: {}", node_id, e)
finally:
ping_task.cancel()
manager.unregister(conn)
if not manager.has_tunnel(node_id):
await _update_tunnel_status(node_id, connected=False)
await _update_tunnel_status(node_id, connected=False, detail=disconnect_reason)
else:
logger.info("tunnel connection closed but pool still active: node_id={}", node_id)
@@ -174,14 +179,16 @@ async def _ping_loop(conn: TunnelConnection) -> None:
pass
async def _update_tunnel_status(node_id: str, *, connected: bool) -> None:
"""更新 ProxyNode 的 tunnel 连接状态(在线程池中执行,避免阻塞 event loop"""
async def _update_tunnel_status(
node_id: str, *, connected: bool, detail: str | None = None
) -> None:
"""更新 ProxyNode 的 tunnel 连接状态并记录事件(在线程池中执行)"""
def _sync_update() -> None:
from datetime import datetime, timezone
from src.database import create_session
from src.models.database import ProxyNode, ProxyNodeStatus
from src.models.database import ProxyNode, ProxyNodeEvent, ProxyNodeStatus
db = create_session()
try:
@@ -193,9 +200,16 @@ async def _update_tunnel_status(node_id: str, *, connected: bool) -> None:
node.tunnel_connected_at = now
node.status = ProxyNodeStatus.ONLINE
else:
# 记录断开时刻,供 health_scheduler 计算 UNHEALTHY 缓冲期
node.tunnel_connected_at = now
node.status = ProxyNodeStatus.UNHEALTHY
node.status = ProxyNodeStatus.OFFLINE
# 记录连接事件
event = ProxyNodeEvent(
node_id=node_id,
event_type="connected" if connected else "disconnected",
detail=detail,
)
db.add(event)
db.commit()
finally:
db.close()

View File

@@ -131,7 +131,7 @@ class ClaudeChatAdapter(ChatAdapterBase):
request_body: dict[str, Any] | None = None,
) -> dict[str, bool]:
"""检测 Claude 请求中隐含的能力需求"""
return ClaudeCapabilityDetector.detect_from_headers(headers)
return ClaudeCapabilityDetector.detect_from_headers(headers, request_body)
# =========================================================================
# Claude 特定的计费逻辑

View File

@@ -874,7 +874,6 @@ class ProxyNodeStatus(PyEnum):
"""代理节点状态"""
ONLINE = "online"
UNHEALTHY = "unhealthy"
OFFLINE = "offline"
@@ -919,6 +918,9 @@ class ProxyNode(Base):
active_connections = Column(Integer, default=0, nullable=False)
total_requests = Column(BigInteger, default=0, nullable=False)
avg_latency_ms = Column(Float, nullable=True)
failed_requests = Column(BigInteger, default=0, nullable=False, comment="累计失败请求数")
dns_failures = Column(BigInteger, default=0, nullable=False, comment="累计 DNS 失败数")
stream_errors = Column(BigInteger, default=0, nullable=False, comment="累计流错误数")
# 硬件信息注册时上报JSON 可扩展)
hardware_info = Column(
@@ -962,6 +964,33 @@ class ProxyNode(Base):
__table_args__ = (UniqueConstraint("ip", "port", name="uq_proxy_node_ip_port"),)
class ProxyNodeEvent(Base):
"""代理节点连接事件表 -- 记录 tunnel 连接/断开/错误事件,用于连接稳定性分析"""
__tablename__ = "proxy_node_events"
id = Column(BigInteger, primary_key=True, autoincrement=True)
node_id = Column(
String(36),
ForeignKey("proxy_nodes.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
event_type = Column(
String(20),
nullable=False,
comment="事件类型: connected, disconnected, error",
)
detail = Column(String(500), nullable=True, comment="事件详情(如断开原因)")
created_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
nullable=False,
)
__table_args__ = (Index("idx_proxy_node_events_node_created", "node_id", "created_at"),)
class GlobalModel(ExportMixin, Base):
"""全局统一模型定义 - 包含价格和能力配置

View File

@@ -21,7 +21,7 @@ if TYPE_CHECKING:
def _reset_tunnel_connected_on_startup() -> None:
"""服务端启动时将所有 tunnel_connected=True 的节点重置为 False/UNHEALTHY
"""服务端启动时将所有 tunnel_connected=True 的节点重置为 False/OFFLINE
服务端重启后 TunnelManager 内存状态丢失,但 DB 中可能残留
tunnel_connected=True 的记录。如果不重置health_scheduler 会错误地
@@ -48,7 +48,7 @@ def _reset_tunnel_connected_on_startup() -> None:
for node in stale_nodes:
node.tunnel_connected = False
node.tunnel_connected_at = now
node.status = ProxyNodeStatus.UNHEALTHY
node.status = ProxyNodeStatus.OFFLINE
node.updated_at = now
db.commit()
logger.info(

View File

@@ -224,15 +224,32 @@ _VALID_CACHE_TTL_TARGETS = {"ephemeral", "1h"}
def _override_cache_control_in_blocks(blocks: list[Any], target: str) -> int:
"""Override cache_control in a list of content blocks. Returns count of overrides."""
"""Override cache_control TTL in a list of content blocks. Returns count of overrides.
According to Anthropic API docs, cache_control format is:
{"type": "ephemeral", "ttl": "5m" | "1h"}
``type`` is always "ephemeral"; the ``ttl`` field controls the actual duration.
When ttl is absent, the default is 5m (ephemeral).
"""
count = 0
for block in blocks:
if not isinstance(block, dict):
continue
cc = block.get("cache_control")
if isinstance(cc, dict):
if cc.get("type") != target:
cc["type"] = target
if not isinstance(cc, dict):
continue
# Ensure type is always "ephemeral"
if cc.get("type") != "ephemeral":
cc["type"] = "ephemeral"
if target == "ephemeral":
# Target is 5m (default) -- remove explicit ttl so it falls back to default
if "ttl" in cc:
del cc["ttl"]
count += 1
else:
# Target is "1h" -- set ttl explicitly
if cc.get("ttl") != target:
cc["ttl"] = target
count += 1
return count

View File

@@ -2,9 +2,9 @@
ProxyNode 心跳检测调度器
定期检查 proxy_nodes 的 tunnel 连接状态,更新节点状态:
- tunnel 实际连接中 -> ONLINE(包括从 OFFLINE 恢复的情况)
- tunnel 刚断开 (<60s) -> UNHEALTHY缓冲期避免正在进行的请求被立即切走
- tunnel 断开超过 60s -> OFFLINE
- tunnel 实际连接中 -> ONLINE
- tunnel 未连接 -> OFFLINE
以 TunnelManager 内存中的实际连接状态为准。
"""
from __future__ import annotations
@@ -17,12 +17,18 @@ from src.database import create_session
from src.models.database import ProxyNode, ProxyNodeStatus
from src.services.system.scheduler import get_scheduler
# 事件保留天数
_EVENT_RETENTION_DAYS = 30
# 每隔多少次心跳检测执行一次事件清理15s * 240 = 1h
_EVENT_CLEANUP_INTERVAL = 240
class ProxyNodeHealthScheduler:
"""代理节点心跳检测调度器"""
def __init__(self) -> None:
self.running = False
self._check_count = 0
async def start(self) -> Any:
if self.running:
@@ -51,6 +57,9 @@ class ProxyNodeHealthScheduler:
async def _scheduled_check(self) -> None:
await self._check_heartbeats()
self._check_count = (self._check_count + 1) % _EVENT_CLEANUP_INTERVAL
if self._check_count == 0:
await self._cleanup_old_events()
async def _check_heartbeats(self) -> None:
from src.services.proxy_node.tunnel_manager import get_tunnel_manager
@@ -87,16 +96,9 @@ class ProxyNodeHealthScheduler:
node.tunnel_connected_at = now
changed += 1
if actually_connected:
new_status = ProxyNodeStatus.ONLINE
elif node.tunnel_connected_at:
# tunnel 刚断开:给 60s 缓冲期标记为 UNHEALTHY
elapsed = (now - node.tunnel_connected_at).total_seconds()
new_status = (
ProxyNodeStatus.UNHEALTHY if elapsed < 60 else ProxyNodeStatus.OFFLINE
)
else:
new_status = ProxyNodeStatus.OFFLINE
new_status = (
ProxyNodeStatus.ONLINE if actually_connected else ProxyNodeStatus.OFFLINE
)
if node.status != new_status:
node.status = new_status
@@ -115,6 +117,42 @@ class ProxyNodeHealthScheduler:
finally:
db.close()
async def _cleanup_old_events(self) -> None:
"""清理超过保留期的连接事件记录(在线程池中执行,避免阻塞事件循环)"""
import asyncio
def _sync_cleanup() -> None:
from datetime import timedelta
from src.models.database import ProxyNodeEvent
db = create_session()
try:
cutoff = datetime.now(timezone.utc) - timedelta(days=_EVENT_RETENTION_DAYS)
deleted = (
db.query(ProxyNodeEvent)
.filter(ProxyNodeEvent.created_at < cutoff)
.delete(synchronize_session=False)
)
if deleted:
db.commit()
logger.info(
"清理 {} 条过期代理节点事件 (>{} 天)", deleted, _EVENT_RETENTION_DAYS
)
except Exception as e:
try:
db.rollback()
except Exception:
pass
logger.warning("清理代理节点事件失败: {}", e)
finally:
db.close()
try:
await asyncio.to_thread(_sync_cleanup)
except Exception as e:
logger.warning("清理代理节点事件线程执行失败: {}", e)
_proxy_node_health_scheduler: ProxyNodeHealthScheduler | None = None

View File

@@ -60,17 +60,39 @@ def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
db = create_session()
try:
node = db.query(ProxyNode).filter(ProxyNode.id == node_id).first()
if not node or node.status != ProxyNodeStatus.ONLINE:
if not node:
_proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS)
return None
# tunnel 模式节点必须 tunnel 已连接才可用
if node.tunnel_mode and not node.tunnel_connected:
# tunnel 模式节点:以 TunnelManager 内存中的实际连接状态为准,
# 而非依赖 DB 的 status/tunnel_connected 字段(可能因竞态不同步)。
if node.tunnel_mode and not node.is_manual:
from src.services.proxy_node.tunnel_manager import get_tunnel_manager
manager = get_tunnel_manager()
if not manager.has_tunnel(node_id):
_proxy_node_cache[node_id] = (
None,
now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS,
)
return None
value: dict[str, Any] = {
"name": node.name,
"ip": node.ip,
"port": node.port,
"tunnel_mode": True,
"tunnel_connected": True,
}
_proxy_node_cache[node_id] = (value, now + _PROXY_NODE_CACHE_TTL_SECONDS)
return value
# 手动节点 / 非 tunnel 节点:仍依赖 DB status
if node.status != ProxyNodeStatus.ONLINE:
_proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS)
return None
if node.is_manual:
value: dict[str, Any] = {
value = {
"is_manual": True,
"name": node.name,
"proxy_url": node.proxy_url,

View File

@@ -58,6 +58,9 @@ def node_to_dict(node: ProxyNode) -> dict[str, Any]:
"active_connections": node.active_connections,
"total_requests": node.total_requests,
"avg_latency_ms": node.avg_latency_ms,
"failed_requests": node.failed_requests,
"dns_failures": node.dns_failures,
"stream_errors": node.stream_errors,
"hardware_info": node.hardware_info,
"estimated_max_concurrency": node.estimated_max_concurrency,
"remote_config": node.remote_config,
@@ -283,7 +286,7 @@ class ProxyNodeService:
port=port,
region=region,
# 新节点:等 tunnel 连接后才上线
status=ProxyNodeStatus.UNHEALTHY,
status=ProxyNodeStatus.OFFLINE,
registered_by=registered_by,
last_heartbeat_at=now,
heartbeat_interval=heartbeat_interval,
@@ -311,8 +314,16 @@ class ProxyNodeService:
active_connections: int | None = None,
total_requests: int | None = None,
avg_latency_ms: float | None = None,
failed_requests: int | None = None,
dns_failures: int | None = None,
stream_errors: int | None = None,
) -> ProxyNode:
"""处理节点心跳(仅 tunnel 模式节点,更新指标并修正状态不一致)"""
"""处理节点心跳(仅 tunnel 模式节点,更新指标并修正状态不一致)
注意: total_requests / failed_requests / dns_failures / stream_errors
来自 Rust 端的区间增量swap(0) 后上报),需要累加到 DB 而非覆盖。
active_connections 和 avg_latency_ms 是实时快照,直接覆盖。
"""
node = db.query(ProxyNode).filter(ProxyNode.id == node_id).first()
if not node:
raise NotFoundException(f"ProxyNode {node_id} 不存在", "proxy_node")
@@ -332,13 +343,23 @@ class ProxyNodeService:
node.last_heartbeat_at = now
if heartbeat_interval is not None:
node.heartbeat_interval = heartbeat_interval
# 实时快照指标 -- 直接覆盖
if active_connections is not None:
node.active_connections = active_connections
if total_requests is not None:
node.total_requests = total_requests
if avg_latency_ms is not None:
node.avg_latency_ms = avg_latency_ms
# 区间增量指标 -- 累加到累计值
if total_requests is not None and total_requests > 0:
node.total_requests = (node.total_requests or 0) + total_requests
if failed_requests is not None and failed_requests > 0:
node.failed_requests = (node.failed_requests or 0) + failed_requests
if dns_failures is not None and dns_failures > 0:
node.dns_failures = (node.dns_failures or 0) + dns_failures
if stream_errors is not None and stream_errors > 0:
node.stream_errors = (node.stream_errors or 0) + stream_errors
db.commit()
db.refresh(node)
return node
@@ -530,7 +551,12 @@ class ProxyNodeService:
# tunnel 节点:通过 WebSocket tunnel 测试
if not node.is_manual:
if not node.tunnel_connected:
# 以 TunnelManager 内存中的实际连接状态为准(与 health_scheduler 一致),
# 而非仅依赖 DB 的 tunnel_connected 字段,避免竞态导致误判。
from src.services.proxy_node.tunnel_manager import get_tunnel_manager
manager = get_tunnel_manager()
if not manager.has_tunnel(node.id):
return {
"success": False,
"latency_ms": None,

View File

@@ -402,6 +402,9 @@ class TunnelManager:
active_connections=data.get("active_connections"),
total_requests=data.get("total_requests"),
avg_latency_ms=data.get("avg_latency_ms"),
failed_requests=data.get("failed_requests"),
dns_failures=data.get("dns_failures"),
stream_errors=data.get("stream_errors"),
)
result: dict[str, Any] = {}
if node.remote_config: