Sakurasan
d4c5263f1e
运行期冷却只存在网关进程内存里，DB 的 health_status 看不见它：渠道被踢到 30s 冷却后，管理后台仍显示 healthy，模型也照常出现在 /v1/models，但请求 返回 503 no_channel，管理员只能去翻日志。 - channel.Service 增 RuntimeState（连续失败次数 + 冷却剩余）与 ClearRuntime； Gateway 包装出 ChannelRuntime / RecoverChannel 供管理端调用。 - GET /api/v1/admin/channels 每项返回 runtime_fails 与 cooldown_seconds。 - 新增 POST /api/v1/admin/channels/:id/recover：清运行期冷却与失败计数，并把 探测写的 health_status 置回 healthy；不改 enabled，上游若真挂着，下一次探测 或再连续 3 次请求失败会重新把它踢掉。 - 渠道页健康列下方显示「冷却中 Ns」/「连续失败 n/3」，仅在冷却或探测不健康时 出现「立即恢复」按钮（卡片与桌面表格两套布局都覆盖）。 仍未处理：冷却到期是裸恢复、没有 half-open 试探；计数是进程内的，多实例各自 为战，要共享得等 Redis 那一步。
432 lines
13 KiB
Go
432 lines
13 KiB
Go
package api
|
||
|
||
import (
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
"github.com/openteam/server/internal/pkg/resp"
|
||
"github.com/openteam/server/internal/store"
|
||
)
|
||
|
||
// AdminChannels GET /api/v1/admin/channels — 渠道列表（不返回加密 key，返回掩码）。
|
||
func (h *Handler) AdminChannels(c *gin.Context) {
|
||
var chs []store.Channel
|
||
if err := h.a.DB.Order("id ASC").Find(&chs).Error; err != nil {
|
||
resp.Fail(c, http.StatusInternalServerError, "failed to load channels")
|
||
return
|
||
}
|
||
out := make([]gin.H, 0, len(chs))
|
||
for _, ch := range chs {
|
||
masked := ""
|
||
if key, err := h.a.Enc.Decrypt(ch.APIKeyEnc); err == nil && len(key) > 8 {
|
||
masked = maskAPIKey(key)
|
||
} else if err == nil {
|
||
masked = "****"
|
||
}
|
||
fails, cooldownSec := h.channelRuntime(ch.ID)
|
||
out = append(out, gin.H{
|
||
"id": ch.ID, "name": ch.Name, "provider": ch.Provider, "formats": ch.FormatsEffective(),
|
||
"base_url": ch.BaseURL, "base_urls": ch.BaseURLs,
|
||
"api_key_masked": masked, "weight": ch.Weight, "priority": ch.Priority,
|
||
"timeout_ms": ch.TimeoutMS, "max_concurrency": ch.MaxConcurrency,
|
||
"health_status": ch.HealthStatus, "enabled": ch.Enabled,
|
||
"runtime_fails": fails, "cooldown_seconds": cooldownSec,
|
||
"created_at": ch.CreatedAt,
|
||
})
|
||
}
|
||
resp.OK(c, gin.H{"items": out})
|
||
}
|
||
|
||
// channelRuntime 渠道运行期健康：连续失败次数与冷却剩余秒数（0 表示无冷却）。
|
||
// 运行期冷却只存在于进程内存，DB 的 health_status 看不到它，故列表单独返回。
|
||
func (h *Handler) channelRuntime(id uint64) (fails int, cooldownSeconds int64) {
|
||
if h.gw == nil {
|
||
return 0, 0
|
||
}
|
||
fails, remaining := h.gw.ChannelRuntime(id)
|
||
return fails, int64((remaining + time.Second - time.Nanosecond) / time.Second) // 向上取整
|
||
}
|
||
|
||
type channelBody struct {
|
||
Name string `json:"name" binding:"required,min=1,max=64"`
|
||
Provider string `json:"provider"` // 可选：为空时按 formats 推断（兼容旧数据）
|
||
Formats []string `json:"formats"` // 原生支持的协议 chat|responses|messages（主配置）
|
||
BaseURL string `json:"base_url"` // 可选：留空按供应商默认；支持前缀或完整端点
|
||
BaseURLs map[string]string `json:"base_urls"` // 分协议 base_url 覆盖（chat/responses/messages）
|
||
APIKey string `json:"api_key"`
|
||
Weight *int `json:"weight"`
|
||
Priority *int `json:"priority"`
|
||
TimeoutMS *int `json:"timeout_ms"`
|
||
MaxConcurrency *int `json:"max_concurrency"`
|
||
Enabled *bool `json:"enabled"`
|
||
}
|
||
|
||
// normalizeBaseURLs 校验并清理分协议 base_url。
|
||
func normalizeBaseURLs(m map[string]string) map[string]string {
|
||
if len(m) == 0 {
|
||
return nil
|
||
}
|
||
out := map[string]string{}
|
||
for k, v := range m {
|
||
if validFormats[k] && strings.TrimSpace(v) != "" {
|
||
out[k] = strings.TrimRight(strings.TrimSpace(v), "/")
|
||
}
|
||
}
|
||
if len(out) == 0 {
|
||
return nil
|
||
}
|
||
return out
|
||
}
|
||
|
||
// resolveBaseURL 渠道 base_url：留空按供应商默认；网关按内容智能识别前缀/完整端点。
|
||
func resolveBaseURL(provider, raw string) (string, error) {
|
||
base := strings.TrimRight(raw, "/")
|
||
if base == "" {
|
||
switch provider {
|
||
case store.ChannelProviderOpenAI:
|
||
base = "https://api.openai.com"
|
||
case store.ChannelProviderAnthropic:
|
||
base = "https://api.anthropic.com"
|
||
}
|
||
}
|
||
if base == "" {
|
||
return "", errors.New("base_url required for compatible channels")
|
||
}
|
||
return base, nil
|
||
}
|
||
|
||
func validateProvider(p string) bool {
|
||
return p == store.ChannelProviderOpenAI || p == store.ChannelProviderAnthropic || p == store.ChannelProviderCompatible
|
||
}
|
||
|
||
var validFormats = map[string]bool{
|
||
store.FormatChat: true, store.FormatResponses: true, store.FormatMessages: true,
|
||
}
|
||
|
||
// deriveProvider 按格式推断供应商（仅作内部字段/兼容用途，不参与路由）。
|
||
func deriveProvider(formats []string) string {
|
||
if len(formats) == 0 {
|
||
return store.ChannelProviderCompatible
|
||
}
|
||
messagesOnly, hasResponses := true, false
|
||
for _, f := range formats {
|
||
if f != store.FormatMessages {
|
||
messagesOnly = false
|
||
}
|
||
if f == store.FormatResponses {
|
||
hasResponses = true
|
||
}
|
||
}
|
||
if messagesOnly {
|
||
return store.ChannelProviderAnthropic
|
||
}
|
||
if hasResponses {
|
||
return store.ChannelProviderOpenAI
|
||
}
|
||
return store.ChannelProviderCompatible
|
||
}
|
||
|
||
// resolveFormats 渠道协议格式：显式给出则校验去重；空则按 provider 推断默认。
|
||
func resolveFormats(provider string, formats []string) ([]string, error) {
|
||
if len(formats) == 0 {
|
||
switch provider {
|
||
case store.ChannelProviderAnthropic:
|
||
return []string{store.FormatMessages}, nil
|
||
case store.ChannelProviderOpenAI:
|
||
return []string{store.FormatChat, store.FormatResponses}, nil
|
||
default:
|
||
return []string{store.FormatChat}, nil
|
||
}
|
||
}
|
||
seen := map[string]bool{}
|
||
out := make([]string, 0, len(formats))
|
||
for _, f := range formats {
|
||
if !validFormats[f] {
|
||
return nil, fmt.Errorf("unsupported format %q", f)
|
||
}
|
||
if !seen[f] {
|
||
seen[f] = true
|
||
out = append(out, f)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// AdminCreateChannel POST /api/v1/admin/channels
|
||
func (h *Handler) AdminCreateChannel(c *gin.Context) {
|
||
var req channelBody
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, "invalid input: "+err.Error())
|
||
return
|
||
}
|
||
if req.Provider == "" {
|
||
req.Provider = deriveProvider(req.Formats)
|
||
}
|
||
if !validateProvider(req.Provider) {
|
||
resp.Fail(c, http.StatusBadRequest, "provider must be openai, anthropic or compatible")
|
||
return
|
||
}
|
||
if req.APIKey == "" {
|
||
resp.Fail(c, http.StatusBadRequest, "api_key required")
|
||
return
|
||
}
|
||
formats, err := resolveFormats(req.Provider, req.Formats)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
baseURL, err := resolveBaseURL(req.Provider, req.BaseURL)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
enc, err := h.a.Enc.Encrypt(req.APIKey)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusInternalServerError, "failed to encrypt api key")
|
||
return
|
||
}
|
||
ch := store.Channel{
|
||
Name: req.Name, Provider: req.Provider, Formats: formats, BaseURL: baseURL,
|
||
BaseURLs: normalizeBaseURLs(req.BaseURLs),
|
||
APIKeyEnc: enc, Weight: intOr(req.Weight, 1), Priority: intOr(req.Priority, 0),
|
||
TimeoutMS: intOr(req.TimeoutMS, 300000), MaxConcurrency: intOr(req.MaxConcurrency, 16),
|
||
HealthStatus: store.ChannelHealthHealthy, Enabled: boolOr(req.Enabled, true),
|
||
}
|
||
if err := h.a.DB.Create(&ch).Error; err != nil {
|
||
resp.Fail(c, http.StatusConflict, "failed to create channel (name may already exist)")
|
||
return
|
||
}
|
||
resp.Created(c, gin.H{"id": ch.ID, "name": ch.Name})
|
||
}
|
||
|
||
// AdminUpdateChannel PUT /api/v1/admin/channels/:id
|
||
func (h *Handler) AdminUpdateChannel(c *gin.Context) {
|
||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, "invalid channel id")
|
||
return
|
||
}
|
||
var body struct {
|
||
Name *string `json:"name"`
|
||
Provider *string `json:"provider"`
|
||
Formats *[]string `json:"formats"`
|
||
BaseURL *string `json:"base_url"`
|
||
BaseURLs *map[string]string `json:"base_urls"`
|
||
APIKey *string `json:"api_key"`
|
||
Weight *int `json:"weight"`
|
||
Priority *int `json:"priority"`
|
||
TimeoutMS *int `json:"timeout_ms"`
|
||
MaxConcurrency *int `json:"max_concurrency"`
|
||
HealthStatus *string `json:"health_status"`
|
||
Enabled *bool `json:"enabled"`
|
||
}
|
||
if err := c.ShouldBindJSON(&body); err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, "invalid input")
|
||
return
|
||
}
|
||
var ch store.Channel
|
||
if err := h.a.DB.First(&ch, id).Error; err != nil {
|
||
resp.Fail(c, http.StatusNotFound, "channel not found")
|
||
return
|
||
}
|
||
updates := map[string]any{}
|
||
if body.Name != nil {
|
||
updates["name"] = *body.Name
|
||
}
|
||
if body.Provider != nil {
|
||
if !validateProvider(*body.Provider) {
|
||
resp.Fail(c, http.StatusBadRequest, "provider must be openai, anthropic or compatible")
|
||
return
|
||
}
|
||
updates["provider"] = *body.Provider
|
||
}
|
||
if body.BaseURL != nil {
|
||
prov := ch.Provider
|
||
if body.Provider != nil {
|
||
prov = *body.Provider
|
||
}
|
||
b, berr := resolveBaseURL(prov, *body.BaseURL)
|
||
if berr != nil {
|
||
resp.Fail(c, http.StatusBadRequest, berr.Error())
|
||
return
|
||
}
|
||
updates["base_url"] = b
|
||
}
|
||
if body.BaseURLs != nil {
|
||
// base_urls 是 jsonb：手动序列化
|
||
raw, _ := json.Marshal(normalizeBaseURLs(*body.BaseURLs))
|
||
updates["base_urls"] = string(raw)
|
||
}
|
||
if body.APIKey != nil && *body.APIKey != "" {
|
||
enc, err := h.a.Enc.Encrypt(*body.APIKey)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusInternalServerError, "failed to encrypt api key")
|
||
return
|
||
}
|
||
updates["api_key_enc"] = enc
|
||
}
|
||
if body.Weight != nil {
|
||
updates["weight"] = *body.Weight
|
||
}
|
||
if body.Priority != nil {
|
||
updates["priority"] = *body.Priority
|
||
}
|
||
if body.TimeoutMS != nil {
|
||
updates["timeout_ms"] = *body.TimeoutMS
|
||
}
|
||
if body.MaxConcurrency != nil {
|
||
updates["max_concurrency"] = *body.MaxConcurrency
|
||
}
|
||
if body.HealthStatus != nil {
|
||
updates["health_status"] = *body.HealthStatus
|
||
}
|
||
if body.Enabled != nil {
|
||
updates["enabled"] = *body.Enabled
|
||
}
|
||
if body.Formats != nil {
|
||
prov := ch.Provider
|
||
if body.Provider != nil {
|
||
prov = *body.Provider
|
||
}
|
||
formats, ferr := resolveFormats(prov, *body.Formats)
|
||
if ferr != nil {
|
||
resp.Fail(c, http.StatusBadRequest, ferr.Error())
|
||
return
|
||
}
|
||
// formats 是 jsonb：手动序列化为 JSON 字符串（map 更新不走序列化）
|
||
raw, _ := json.Marshal(formats)
|
||
updates["formats"] = string(raw)
|
||
}
|
||
if len(updates) > 0 {
|
||
if err := h.a.DB.Model(&ch).Updates(updates).Error; err != nil {
|
||
resp.Fail(c, http.StatusInternalServerError, "failed to update channel")
|
||
return
|
||
}
|
||
}
|
||
resp.OK(c, gin.H{"ok": true})
|
||
}
|
||
|
||
// AdminDeleteChannel DELETE /api/v1/admin/channels/:id
|
||
func (h *Handler) AdminDeleteChannel(c *gin.Context) {
|
||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, "invalid channel id")
|
||
return
|
||
}
|
||
res := h.a.DB.Delete(&store.Channel{}, id)
|
||
if res.Error != nil {
|
||
resp.Fail(c, http.StatusInternalServerError, "failed to delete channel")
|
||
return
|
||
}
|
||
if res.RowsAffected == 0 {
|
||
resp.Fail(c, http.StatusNotFound, "channel not found")
|
||
return
|
||
}
|
||
// 清理模型绑定
|
||
h.a.DB.Where("channel_id = ?", id).Delete(&store.ChannelModelBinding{})
|
||
resp.OK(c, gin.H{"ok": true})
|
||
}
|
||
|
||
// AdminTestChannel POST /api/v1/admin/channels/:id/test — 请求渠道 /v1/models 测连通性。
|
||
func (h *Handler) AdminTestChannel(c *gin.Context) {
|
||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, "invalid channel id")
|
||
return
|
||
}
|
||
var ch store.Channel
|
||
if err := h.a.DB.First(&ch, id).Error; err != nil {
|
||
resp.Fail(c, http.StatusNotFound, "channel not found")
|
||
return
|
||
}
|
||
key, err := h.a.Enc.Decrypt(ch.APIKeyEnc)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusInternalServerError, "failed to decrypt channel key")
|
||
return
|
||
}
|
||
url := ch.UpstreamURL("", "/models")
|
||
client := &http.Client{Timeout: 10 * time.Second}
|
||
req, _ := http.NewRequest(http.MethodGet, url, nil)
|
||
req.Header.Set("Authorization", "Bearer "+key)
|
||
req.Header.Set("Accept", "application/json")
|
||
start := time.Now()
|
||
resp2, err := client.Do(req)
|
||
status := store.ChannelHealthHealthy
|
||
msg := "ok"
|
||
latency := 0
|
||
if err != nil {
|
||
status = store.ChannelHealthCooldown
|
||
msg = err.Error()
|
||
} else {
|
||
latency = int(time.Since(start).Milliseconds())
|
||
if resp2.StatusCode < 200 || resp2.StatusCode >= 300 {
|
||
status = store.ChannelHealthCooldown
|
||
b, _ := io.ReadAll(io.LimitReader(resp2.Body, 1024))
|
||
msg = fmt.Sprintf("http %d: %s", resp2.StatusCode, strings.TrimSpace(string(b)))
|
||
}
|
||
resp2.Body.Close()
|
||
}
|
||
h.a.DB.Model(&store.Channel{}).Where("id = ?", ch.ID).Update("health_status", status)
|
||
if status != store.ChannelHealthHealthy {
|
||
resp.Fail(c, http.StatusBadGateway, msg)
|
||
return
|
||
}
|
||
resp.OK(c, gin.H{"ok": true, "latency_ms": latency, "message": msg})
|
||
}
|
||
|
||
// AdminRecoverChannel POST /api/v1/admin/channels/:id/recover — 立即恢复渠道：
|
||
// 清掉运行期冷却与连续失败计数，并把探测写的 health_status 置回 healthy。
|
||
// 不改 enabled；上游若真挂着，下一次探测或连续 3 次请求失败会重新把它踢掉。
|
||
func (h *Handler) AdminRecoverChannel(c *gin.Context) {
|
||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||
if err != nil {
|
||
resp.Fail(c, http.StatusBadRequest, "invalid channel id")
|
||
return
|
||
}
|
||
var ch store.Channel
|
||
if err := h.a.DB.First(&ch, id).Error; err != nil {
|
||
resp.Fail(c, http.StatusNotFound, "channel not found")
|
||
return
|
||
}
|
||
if h.gw != nil {
|
||
h.gw.RecoverChannel(ch.ID)
|
||
}
|
||
if err := h.a.DB.Model(&store.Channel{}).Where("id = ?", ch.ID).
|
||
Update("health_status", store.ChannelHealthHealthy).Error; err != nil {
|
||
resp.Fail(c, http.StatusInternalServerError, "failed to update channel")
|
||
return
|
||
}
|
||
resp.OK(c, gin.H{"id": ch.ID, "name": ch.Name,
|
||
"health_status": store.ChannelHealthHealthy, "runtime_fails": 0, "cooldown_seconds": 0})
|
||
}
|
||
|
||
// maskAPIKey 掩码渠道密钥：保留前 7 位与后 4 位，中间固定 ****** 遮蔽。
|
||
// 示例：xxxxxxx******Mq4Y；密钥较短时退化为仅保留后 4 位。
|
||
func maskAPIKey(key string) string {
|
||
if len(key) <= 11 {
|
||
return strings.Repeat("*", len(key)-4) + key[len(key)-4:]
|
||
}
|
||
return key[:7] + "******" + key[len(key)-4:]
|
||
}
|
||
|
||
func intOr(p *int, def int) int {
|
||
if p == nil {
|
||
return def
|
||
}
|
||
return *p
|
||
}
|
||
|
||
func boolOr(p *bool, def bool) bool {
|
||
if p == nil {
|
||
return def
|
||
}
|
||
return *p
|
||
}
|