发布 0.2.0:模型池自愈、探测修正、任务异步触发与删除加固
This commit is contained in:
@@ -23,7 +23,7 @@ jobs:
|
|||||||
echo "IMAGE=${REGISTRY}/${{ gitea.repository }}" >> "$GITHUB_ENV"
|
echo "IMAGE=${REGISTRY}/${{ gitea.repository }}" >> "$GITHUB_ENV"
|
||||||
echo "BUILD_TIME=$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$GITHUB_ENV"
|
echo "BUILD_TIME=$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$GITHUB_ENV"
|
||||||
|
|
||||||
- name: 下载 DASH_VERSION 固定版本的前端 dist.zip,校验后解压进嵌入目录
|
- name: 下载前端 dist.zip
|
||||||
env:
|
env:
|
||||||
TOKEN: ${{ secrets.BUILD_TOKEN }}
|
TOKEN: ${{ secrets.BUILD_TOKEN }}
|
||||||
run: |
|
run: |
|
||||||
@@ -36,7 +36,7 @@ jobs:
|
|||||||
unzip -q dist.zip -d internal/webui/dist
|
unzip -q dist.zip -d internal/webui/dist
|
||||||
test -f internal/webui/dist/index.html
|
test -f internal/webui/dist/index.html
|
||||||
|
|
||||||
- name: 构建双架构二进制(artifact stage 导出)
|
- name: 构建双架构二进制
|
||||||
run: |
|
run: |
|
||||||
docker buildx build --pull \
|
docker buildx build --pull \
|
||||||
--platform linux/amd64,linux/arm64 \
|
--platform linux/amd64,linux/arm64 \
|
||||||
|
|||||||
@@ -2,6 +2,24 @@
|
|||||||
|
|
||||||
格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/),版本号遵循语义化版本。
|
格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/),版本号遵循语义化版本。
|
||||||
|
|
||||||
|
## [0.2.0] - 2026-07-10
|
||||||
|
|
||||||
|
### Added
|
||||||
|
|
||||||
|
- AI 网关模型池自愈:同步与每日探测后自动逐个试调验证模型可用性(maxTokens=1),不可按需调用的模型(微调基座 / 区域未开放按需供给)自动标记并从模型列表、路由与探测候选中剔除;已标记模型每 20 小时复检,OCI 恢复供给后自动解除标记。`ai_model_caches` 新增 `unusable` / `unusable_reason` / `checked_at`(启动自动迁移),渠道详情接口保留标记行与原因便于排查
|
||||||
|
|
||||||
|
### Changed
|
||||||
|
|
||||||
|
- 后台任务「立即执行」改异步触发:接口立即返回 202,执行结果经任务日志轮询呈现;同一任务在途时重复触发返回 409「任务正在执行中」,与 cron 重叠触发静默跳过(此前同步阻塞数十秒且同一任务可并发重复执行)
|
||||||
|
- 渠道探测候选跨厂商分散(上限 8,voice 等不可对话形态排除):单一厂商在区域内全为微调基座时不再拖垮整个渠道的探测结论
|
||||||
|
- 网关调用(对话 / 流式 / 向量化)遇「模型不可按需调用」类错误(微调基座 400 / 实体不存在 404)自动标记该模型并换渠道重试,且不计入渠道熔断;仅鉴权类错误才判定租户无配额
|
||||||
|
- 同名模型多条目去重优先保留非微调基座条目,降低缓存到不可调用 OCID 的概率
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
|
||||||
|
- 渠道「同步模型成功但探测报错不可用」:法兰克福等区域的微调基座模型占满探测候选导致的误报(探测状态与真实可用性不符的根因)
|
||||||
|
- 租户删除:告警命中清理改子查询,日志事件数万条时不再超出 SQL 绑定变量上限导致删除失败;无法解析的任务 payload 记警告跳过并保留原任务,不再永久阻断租户删除
|
||||||
|
|
||||||
## [0.1.0] - 2026-07-10
|
## [0.1.0] - 2026-07-10
|
||||||
|
|
||||||
### Added
|
### Added
|
||||||
|
|||||||
+10
-3
@@ -5738,7 +5738,7 @@ const docTemplate = `{
|
|||||||
"tags": [
|
"tags": [
|
||||||
"任务与日志回传"
|
"任务与日志回传"
|
||||||
],
|
],
|
||||||
"summary": "立即执行任务",
|
"summary": "立即执行任务(异步触发,结果经任务日志轮询获取)",
|
||||||
"parameters": [
|
"parameters": [
|
||||||
{
|
{
|
||||||
"type": "integer",
|
"type": "integer",
|
||||||
@@ -5749,8 +5749,15 @@ const docTemplate = `{
|
|||||||
}
|
}
|
||||||
],
|
],
|
||||||
"responses": {
|
"responses": {
|
||||||
"200": {
|
"202": {
|
||||||
"description": "OK",
|
"description": "Accepted",
|
||||||
|
"schema": {
|
||||||
|
"type": "object",
|
||||||
|
"additionalProperties": true
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"409": {
|
||||||
|
"description": "任务正在执行中",
|
||||||
"schema": {
|
"schema": {
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"additionalProperties": true
|
"additionalProperties": true
|
||||||
|
|||||||
+10
-3
@@ -5731,7 +5731,7 @@
|
|||||||
"tags": [
|
"tags": [
|
||||||
"任务与日志回传"
|
"任务与日志回传"
|
||||||
],
|
],
|
||||||
"summary": "立即执行任务",
|
"summary": "立即执行任务(异步触发,结果经任务日志轮询获取)",
|
||||||
"parameters": [
|
"parameters": [
|
||||||
{
|
{
|
||||||
"type": "integer",
|
"type": "integer",
|
||||||
@@ -5742,8 +5742,15 @@
|
|||||||
}
|
}
|
||||||
],
|
],
|
||||||
"responses": {
|
"responses": {
|
||||||
"200": {
|
"202": {
|
||||||
"description": "OK",
|
"description": "Accepted",
|
||||||
|
"schema": {
|
||||||
|
"type": "object",
|
||||||
|
"additionalProperties": true
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"409": {
|
||||||
|
"description": "任务正在执行中",
|
||||||
"schema": {
|
"schema": {
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"additionalProperties": true
|
"additionalProperties": true
|
||||||
|
|||||||
+8
-3
@@ -4158,14 +4158,19 @@ paths:
|
|||||||
required: true
|
required: true
|
||||||
type: integer
|
type: integer
|
||||||
responses:
|
responses:
|
||||||
"200":
|
"202":
|
||||||
description: OK
|
description: Accepted
|
||||||
|
schema:
|
||||||
|
additionalProperties: true
|
||||||
|
type: object
|
||||||
|
"409":
|
||||||
|
description: 任务正在执行中
|
||||||
schema:
|
schema:
|
||||||
additionalProperties: true
|
additionalProperties: true
|
||||||
type: object
|
type: object
|
||||||
security:
|
security:
|
||||||
- BearerAuth: []
|
- BearerAuth: []
|
||||||
summary: 立即执行任务
|
summary: 立即执行任务(异步触发,结果经任务日志轮询获取)
|
||||||
tags:
|
tags:
|
||||||
- 任务与日志回传
|
- 任务与日志回传
|
||||||
/api/v1/webhooks/oci-logs/{secret}:
|
/api/v1/webhooks/oci-logs/{secret}:
|
||||||
|
|||||||
+10
-5
@@ -2,6 +2,7 @@ package api
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
|
||||||
@@ -155,10 +156,11 @@ func (h *taskHandler) logs(c *gin.Context) {
|
|||||||
c.JSON(http.StatusOK, logs)
|
c.JSON(http.StatusOK, logs)
|
||||||
}
|
}
|
||||||
|
|
||||||
// @Summary 立即执行任务
|
// @Summary 立即执行任务(异步触发,结果经任务日志轮询获取)
|
||||||
// @Tags 任务与日志回传
|
// @Tags 任务与日志回传
|
||||||
// @Param id path int true "配置 ID"
|
// @Param id path int true "配置 ID"
|
||||||
// @Success 200 {object} map[string]any
|
// @Success 202 {object} map[string]any
|
||||||
|
// @Failure 409 {object} map[string]any "任务正在执行中"
|
||||||
// @Security BearerAuth
|
// @Security BearerAuth
|
||||||
// @Router /api/v1/tasks/{id}/run [post]
|
// @Router /api/v1/tasks/{id}/run [post]
|
||||||
func (h *taskHandler) run(c *gin.Context) {
|
func (h *taskHandler) run(c *gin.Context) {
|
||||||
@@ -166,10 +168,13 @@ func (h *taskHandler) run(c *gin.Context) {
|
|||||||
if !ok {
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
entry, err := h.svc.RunTaskNow(c.Request.Context(), id)
|
if err := h.svc.TriggerTask(c.Request.Context(), id); err != nil {
|
||||||
if err != nil {
|
if errors.Is(err, service.ErrTaskRunning) {
|
||||||
|
c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
|
||||||
|
return
|
||||||
|
}
|
||||||
respondError(c, err)
|
respondError(c, err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
c.JSON(http.StatusOK, entry)
|
c.JSON(http.StatusAccepted, gin.H{"triggered": true})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -309,6 +309,13 @@ type AiModelCache struct {
|
|||||||
// 弃用后仍可调用,直到 RetiredAt(按需推理退役,同步层已剔除过期项)。
|
// 弃用后仍可调用,直到 RetiredAt(按需推理退役,同步层已剔除过期项)。
|
||||||
DeprecatedAt *time.Time `json:"deprecatedAt"`
|
DeprecatedAt *time.Time `json:"deprecatedAt"`
|
||||||
RetiredAt *time.Time `json:"retiredAt"`
|
RetiredAt *time.Time `json:"retiredAt"`
|
||||||
|
// Unusable 标记该模型在此渠道不可按需调用(微调基座/区域未开放按需),
|
||||||
|
// 由探测、真实调用或同步后的后台验证习得:不参与网关列表/路由/探测候选;
|
||||||
|
// 任何同步都按 OCID 结转标记,已标记模型定期复检、恢复供给自动解除。
|
||||||
|
Unusable bool `json:"unusable"`
|
||||||
|
UnusableReason string `gorm:"size:200" json:"unusableReason,omitempty"`
|
||||||
|
// CheckedAt 是最近一次可用性验证时间;NULL 表示从未验证(同步后的后台验证以此为队列)。
|
||||||
|
CheckedAt *time.Time `json:"checkedAt"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// AiCallLog 是网关调用日志:仅元数据与 token 用量,绝不记录 prompt / 响应正文。
|
// AiCallLog 是网关调用日志:仅元数据与 token 用量,绝不记录 prompt / 响应正文。
|
||||||
|
|||||||
@@ -76,3 +76,34 @@ func ServiceStatus(err error) (int, bool) {
|
|||||||
}
|
}
|
||||||
return 0, false
|
return 0, false
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// IsOnDemandUnsupported 识别「模型在该区域仅为微调基座、不支持按需调用」的 400:
|
||||||
|
// OCI 消息形如 "Not allowed to call finetune base model …, use Endpoint: false"。
|
||||||
|
// ListModels 无字段可事先区分,只能在调用报错时识别并换渠道。
|
||||||
|
func IsOnDemandUnsupported(err error) bool {
|
||||||
|
var svcErr common.ServiceError
|
||||||
|
if !errors.As(err, &svcErr) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return svcErr.GetHTTPStatusCode() == 400 &&
|
||||||
|
strings.Contains(strings.ToLower(svcErr.GetMessage()), "finetune base model")
|
||||||
|
}
|
||||||
|
|
||||||
|
// IsEntityNotFound 识别「实体不存在」404(消息 "Entity with key … not found"):
|
||||||
|
// GenAI 对区域内无按需供给的模型 OCID 返回此类 404,属模型级错误;
|
||||||
|
// 鉴权失败的 404 是 NotAuthorizedOrNotFound(消息为 Authorization failed…),不在此列。
|
||||||
|
func IsEntityNotFound(err error) bool {
|
||||||
|
var svcErr common.ServiceError
|
||||||
|
if !errors.As(err, &svcErr) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
msg := strings.ToLower(svcErr.GetMessage())
|
||||||
|
return svcErr.GetHTTPStatusCode() == 404 &&
|
||||||
|
strings.Contains(msg, "entity with key") && strings.Contains(msg, "not found")
|
||||||
|
}
|
||||||
|
|
||||||
|
// IsModelUnavailable 归并「模型×区域不可按需调用」两类报错:微调基座 400 与实体不存在 404;
|
||||||
|
// 命中即应把该 (渠道, 模型) 从池中剔除并换候选/换渠道,而非定论租户配额问题。
|
||||||
|
func IsModelUnavailable(err error) bool {
|
||||||
|
return IsOnDemandUnsupported(err) || IsEntityNotFound(err)
|
||||||
|
}
|
||||||
|
|||||||
@@ -109,3 +109,52 @@ func TestErrorHint(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestIsOnDemandUnsupported(t *testing.T) {
|
||||||
|
ft := fakeServiceError{status: 400, code: "InvalidParameter",
|
||||||
|
message: "Not allowed to call finetune base model ocid1.generativeaimodel.oc1.eu-frankfurt-1.tpel5q, use Endpoint: false"}
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
err error
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"微调基座 400 命中(含包装链)", fmt.Errorf("genai chat: %w", ft), true},
|
||||||
|
{"同消息但非 400 不命中", fakeServiceError{status: 500, code: "InternalError", message: ft.message}, false},
|
||||||
|
{"普通 400 不命中", fakeServiceError{status: 400, code: "InvalidParameter", message: "bad request"}, false},
|
||||||
|
{"非服务端错误不命中", errors.New("finetune base model"), false},
|
||||||
|
}
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
if got := IsOnDemandUnsupported(tt.err); got != tt.want {
|
||||||
|
t.Errorf("IsOnDemandUnsupported() = %v, want %v", got, tt.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestIsModelUnavailable(t *testing.T) {
|
||||||
|
entity404 := fakeServiceError{status: 404, code: "NotFound",
|
||||||
|
message: "Entity with key ocid1.generativeaimodel.oc1.eu-frankfurt-1.2flsfq not found"}
|
||||||
|
auth404 := fakeServiceError{status: 404, code: "NotAuthorizedOrNotFound",
|
||||||
|
message: "Authorization failed or requested resource not found."}
|
||||||
|
ft400 := fakeServiceError{status: 400, code: "InvalidParameter",
|
||||||
|
message: "Not allowed to call finetune base model ocid1.generativeaimodel.oc1..x, use Endpoint: false"}
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
err error
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"实体不存在 404 命中(含包装链)", fmt.Errorf("genai chat: %w", entity404), true},
|
||||||
|
{"鉴权类 404 不命中(仍属租户级)", auth404, false},
|
||||||
|
{"微调基座 400 命中", ft400, true},
|
||||||
|
{"其他 404 消息不命中", fakeServiceError{status: 404, code: "NotFound", message: "route not found"}, false},
|
||||||
|
{"非服务端错误不命中", errors.New("entity with key x not found"), false},
|
||||||
|
}
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
if got := IsModelUnavailable(tt.err); got != tt.want {
|
||||||
|
t.Errorf("IsModelUnavailable() = %v, want %v", got, tt.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+27
-7
@@ -20,7 +20,6 @@ type GenAiModel struct {
|
|||||||
Ocid string `json:"ocid"`
|
Ocid string `json:"ocid"`
|
||||||
Name string `json:"name"`
|
Name string `json:"name"`
|
||||||
Vendor string `json:"vendor"`
|
Vendor string `json:"vendor"`
|
||||||
ChatOnly bool `json:"-"`
|
|
||||||
Caps []string `json:"capabilities"`
|
Caps []string `json:"capabilities"`
|
||||||
// Capability 是网关侧归一能力:CHAT / EMBEDDING(兼具时算 CHAT)
|
// Capability 是网关侧归一能力:CHAT / EMBEDDING(兼具时算 CHAT)
|
||||||
Capability string `json:"capability"`
|
Capability string `json:"capability"`
|
||||||
@@ -71,18 +70,39 @@ func (c *RealClient) ListGenAiModels(ctx context.Context, cred Credentials, regi
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("list genai models: %w", err)
|
return nil, fmt.Errorf("list genai models: %w", err)
|
||||||
}
|
}
|
||||||
seen := map[string]bool{}
|
return dedupGenAiModels(resp.Items, time.Now()), nil
|
||||||
now := time.Now()
|
}
|
||||||
|
|
||||||
|
// dedupGenAiModels 压平并按名称去重;同名多条目时优先保留不含 FINE_TUNE 能力的条目
|
||||||
|
// (微调基座条目在部分区域不支持按需调用,缓存其 OCID 会导致调用 400)。
|
||||||
|
func dedupGenAiModels(items []generativeai.ModelSummary, now time.Time) []GenAiModel {
|
||||||
|
seen := map[string]int{}
|
||||||
var out []GenAiModel
|
var out []GenAiModel
|
||||||
for _, m := range resp.Items {
|
for _, m := range items {
|
||||||
gm, ok := toGenAiModel(m, now)
|
gm, ok := toGenAiModel(m, now)
|
||||||
if !ok || seen[gm.Name] {
|
if !ok {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
seen[gm.Name] = true
|
if i, dup := seen[gm.Name]; dup {
|
||||||
|
if hasFineTune(out[i].Caps) && !hasFineTune(gm.Caps) {
|
||||||
|
out[i] = gm
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seen[gm.Name] = len(out)
|
||||||
out = append(out, gm)
|
out = append(out, gm)
|
||||||
}
|
}
|
||||||
return out, nil
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// hasFineTune 判断能力列表是否含 FINE_TUNE(微调基座条目)。
|
||||||
|
func hasFineTune(caps []string) bool {
|
||||||
|
for _, c := range caps {
|
||||||
|
if c == string(generativeai.ModelCapabilityFineTune) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
// toGenAiModel 压平模型摘要;无归一能力、无名称或按需推理已退役
|
// toGenAiModel 压平模型摘要;无归一能力、无名称或按需推理已退役
|
||||||
|
|||||||
@@ -38,3 +38,33 @@ func TestToGenAiModelRetiredFilter(t *testing.T) {
|
|||||||
t.Error("正常模型应入池")
|
t.Error("正常模型应入池")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDedupGenAiModelsFineTunePreference(t *testing.T) {
|
||||||
|
now := time.Date(2026, 7, 10, 0, 0, 0, 0, time.UTC)
|
||||||
|
chat := []generativeai.ModelCapabilityEnum{generativeai.ModelCapabilityChat}
|
||||||
|
chatFT := []generativeai.ModelCapabilityEnum{generativeai.ModelCapabilityChat, generativeai.ModelCapabilityFineTune}
|
||||||
|
mk := func(id, name string, caps []generativeai.ModelCapabilityEnum) generativeai.ModelSummary {
|
||||||
|
return generativeai.ModelSummary{
|
||||||
|
Id: common.String(id), DisplayName: common.String(name), Vendor: common.String("meta"),
|
||||||
|
Capabilities: caps, LifecycleState: generativeai.ModelLifecycleStateActive,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const name = "meta.llama-3-70b-instruct"
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
items []generativeai.ModelSummary
|
||||||
|
want string // 期望保留的 OCID
|
||||||
|
}{
|
||||||
|
{"基座在前、纯对话在后:保留纯对话条目", []generativeai.ModelSummary{mk("o-ft", name, chatFT), mk("o-od", name, chat)}, "o-od"},
|
||||||
|
{"纯对话在前、基座在后:保留纯对话条目", []generativeai.ModelSummary{mk("o-od", name, chat), mk("o-ft", name, chatFT)}, "o-od"},
|
||||||
|
{"仅基座条目:保留不误删", []generativeai.ModelSummary{mk("o-ft", name, chatFT)}, "o-ft"},
|
||||||
|
}
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
got := dedupGenAiModels(tt.items, now)
|
||||||
|
if len(got) != 1 || got[0].Ocid != tt.want {
|
||||||
|
t.Errorf("dedupGenAiModels() = %+v, want 仅保留 %s", got, tt.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+223
-19
@@ -30,6 +30,10 @@ const (
|
|||||||
aiBackoffCap = 30 * time.Minute
|
aiBackoffCap = 30 * time.Minute
|
||||||
// aiKeyTouchGap 是 LastUsedAt 的最小写库间隔,避免高频调用刷库
|
// aiKeyTouchGap 是 LastUsedAt 的最小写库间隔,避免高频调用刷库
|
||||||
aiKeyTouchGap = time.Minute
|
aiKeyTouchGap = time.Minute
|
||||||
|
// modelRecheckGap 是已标记不可用模型的复检间隔(恢复供给自动解除标记);
|
||||||
|
// validateBatchCap 限制单渠道单轮验证的试调次数
|
||||||
|
modelRecheckGap = 20 * time.Hour
|
||||||
|
validateBatchCap = 32
|
||||||
// 内容日志(红线例外)约束:开启必须限时(上限 7 天),正文截断,短保留
|
// 内容日志(红线例外)约束:开启必须限时(上限 7 天),正文截断,短保留
|
||||||
aiContentLogMaxHours = 168
|
aiContentLogMaxHours = 168
|
||||||
aiContentLogRetention = 7 * 24 * time.Hour
|
aiContentLogRetention = 7 * 24 * time.Hour
|
||||||
@@ -55,13 +59,17 @@ type AiGatewayService struct {
|
|||||||
// touchMu 保护各密钥的最近触达时间(内存节流,不追求跨实例精确)
|
// touchMu 保护各密钥的最近触达时间(内存节流,不追求跨实例精确)
|
||||||
touchMu sync.Mutex
|
touchMu sync.Mutex
|
||||||
lastTouch map[uint]time.Time
|
lastTouch map[uint]time.Time
|
||||||
|
// validateMu 保护 validating:同一渠道的模型验证不并发
|
||||||
|
validateMu sync.Mutex
|
||||||
|
validating map[uint]bool
|
||||||
// onChannelsChanged 在渠道增删后触发,由 main 装配为探测任务同步钩子
|
// onChannelsChanged 在渠道增删后触发,由 main 装配为探测任务同步钩子
|
||||||
onChannelsChanged func(context.Context)
|
onChannelsChanged func(context.Context)
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewAiGatewayService 组装依赖;调用 StartCleanup 后开始调用日志周期清理。
|
// NewAiGatewayService 组装依赖;调用 StartCleanup 后开始调用日志周期清理。
|
||||||
func NewAiGatewayService(db *gorm.DB, configs *OciConfigService, client oci.Client) *AiGatewayService {
|
func NewAiGatewayService(db *gorm.DB, configs *OciConfigService, client oci.Client) *AiGatewayService {
|
||||||
return &AiGatewayService{db: db, configs: configs, client: client, lastTouch: map[uint]time.Time{}}
|
return &AiGatewayService{db: db, configs: configs, client: client,
|
||||||
|
lastTouch: map[uint]time.Time{}, validating: map[uint]bool{}}
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetOnChannelsChanged 注册渠道数量变化钩子(渠道创建/删除成功后调用)。
|
// SetOnChannelsChanged 注册渠道数量变化钩子(渠道创建/删除成功后调用)。
|
||||||
@@ -290,7 +298,7 @@ func (s *AiGatewayService) ProbeChannel(ctx context.Context, id uint) (*model.Ai
|
|||||||
return &fresh, nil
|
return &fresh, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// probe 执行探测并返回 (状态, 错误摘要);同时完成模型缓存同步。
|
// probe 执行探测并返回 (状态, 错误摘要);同时完成模型缓存同步(保留不可用标记)。
|
||||||
func (s *AiGatewayService) probe(ctx context.Context, cred oci.Credentials, ch *model.AiChannel) (string, string) {
|
func (s *AiGatewayService) probe(ctx context.Context, cred oci.Credentials, ch *model.AiChannel) (string, string) {
|
||||||
models, err := s.client.ListGenAiModels(ctx, cred, ch.Region)
|
models, err := s.client.ListGenAiModels(ctx, cred, ch.Region)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -303,43 +311,209 @@ func (s *AiGatewayService) probe(ctx context.Context, cred oci.Credentials, ch *
|
|||||||
if err := s.replaceModels(ctx, ch.ID, models); err != nil {
|
if err := s.replaceModels(ctx, ch.ID, models); err != nil {
|
||||||
return "error", truncateErr(err.Error())
|
return "error", truncateErr(err.Error())
|
||||||
}
|
}
|
||||||
return s.probeChat(ctx, cred, ch, models)
|
usable, err := s.usableModels(ctx, ch.ID, models)
|
||||||
|
if err != nil {
|
||||||
|
return "error", truncateErr(err.Error())
|
||||||
|
}
|
||||||
|
return s.probeChat(ctx, cred, ch, usable)
|
||||||
}
|
}
|
||||||
|
|
||||||
// probeChat 按偏好挑选至多 3 个模型依次试调:部分模型元数据标 CHAT 但实际
|
// usableModels 过滤掉缓存中已标记不可按需调用的模型,探测候选不再反复踩坑。
|
||||||
// 不可对话(如 voice agent),遇 400/5xx 换下一个;401/403/404 属租户级直接定论。
|
func (s *AiGatewayService) usableModels(ctx context.Context, channelID uint, models []oci.GenAiModel) ([]oci.GenAiModel, error) {
|
||||||
|
marks, err := s.loadModelMarks(ctx, channelID)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out := make([]oci.GenAiModel, 0, len(models))
|
||||||
|
for _, m := range models {
|
||||||
|
if !marks[m.Ocid].Unusable {
|
||||||
|
out = append(out, m)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// probeChat 按候选顺序试调(上限 8):遇「模型不可按需调用」标记剔除并换下一个;
|
||||||
|
// 401/403 与鉴权类 404 属租户级直接定论 no_quota;其余错误(部分模型元数据标 CHAT
|
||||||
|
// 但实际不可对话,如 voice agent)累计 3 次止损。
|
||||||
func (s *AiGatewayService) probeChat(ctx context.Context, cred oci.Credentials, ch *model.AiChannel, models []oci.GenAiModel) (string, string) {
|
func (s *AiGatewayService) probeChat(ctx context.Context, cred oci.Credentials, ch *model.AiChannel, models []oci.GenAiModel) (string, string) {
|
||||||
status, detail := "error", "无可试调对话模型"
|
status, detail := "error", "无可试调对话模型"
|
||||||
|
errBudget := 3
|
||||||
for _, m := range probeCandidates(models) {
|
for _, m := range probeCandidates(models) {
|
||||||
code, err := s.client.GenAiProbeChat(ctx, cred, ch.Region, m.Ocid, m.Name)
|
code, err := s.client.GenAiProbeChat(ctx, cred, ch.Region, m.Ocid, m.Name)
|
||||||
switch {
|
switch {
|
||||||
case code == 200 || code == 429:
|
case code == 200 || code == 429:
|
||||||
return "ok", ""
|
return "ok", ""
|
||||||
|
case oci.IsModelUnavailable(err):
|
||||||
|
s.markModelUnusable(ctx, ch.ID, m.Ocid, oci.CompactError(err))
|
||||||
|
status, detail = "error", truncateErr(fmt.Sprintf("%s: 不可按需调用,已从模型池剔除", m.Name))
|
||||||
case code == 401 || code == 403 || code == 404:
|
case code == 401 || code == 403 || code == 404:
|
||||||
return "no_quota", truncateErr(oci.CompactError(err))
|
return "no_quota", truncateErr(oci.CompactError(err))
|
||||||
default:
|
default:
|
||||||
status, detail = "error", truncateErr(fmt.Sprintf("%s: %s", m.Name, oci.CompactError(err)))
|
status, detail = "error", truncateErr(fmt.Sprintf("%s: %s", m.Name, oci.CompactError(err)))
|
||||||
|
if errBudget--; errBudget == 0 {
|
||||||
|
return status, detail
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return status, detail
|
return status, detail
|
||||||
}
|
}
|
||||||
|
|
||||||
// probeCandidates 只取对话模型并按可靠度排序取前 3:主流文本模型优先,
|
// markModelUnusable 把 (渠道, 模型OCID) 标记为不可按需调用;失败仅记日志不阻断主流程。
|
||||||
// voice 等非常规形态殿后(embedding / rerank 已被能力筛选排除)。
|
func (s *AiGatewayService) markModelUnusable(ctx context.Context, channelID uint, ocid, reason string) {
|
||||||
|
err := s.db.WithContext(ctx).Model(&model.AiModelCache{}).
|
||||||
|
Where("channel_id = ? AND model_ocid = ?", channelID, ocid).
|
||||||
|
Updates(map[string]any{"unusable": true, "unusable_reason": shortReason(reason), "checked_at": time.Now()}).Error
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("mark model unusable: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func shortReason(reason string) string {
|
||||||
|
if len(reason) > 200 {
|
||||||
|
return reason[:200]
|
||||||
|
}
|
||||||
|
return reason
|
||||||
|
}
|
||||||
|
|
||||||
|
// beginValidate 抢占渠道的验证执行权,同渠道同时只跑一轮。
|
||||||
|
func (s *AiGatewayService) beginValidate(id uint) bool {
|
||||||
|
s.validateMu.Lock()
|
||||||
|
defer s.validateMu.Unlock()
|
||||||
|
if s.validating[id] {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
s.validating[id] = true
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *AiGatewayService) endValidate(id uint) {
|
||||||
|
s.validateMu.Lock()
|
||||||
|
delete(s.validating, id)
|
||||||
|
s.validateMu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateTargets 取待验证行:从未验证的新模型,以及标记超过复检间隔的模型(仅对话能力,
|
||||||
|
// 兼容存量空串);单轮上限 validateBatchCap 控制调用量。
|
||||||
|
func (s *AiGatewayService) validateTargets(ctx context.Context, channelID uint) ([]model.AiModelCache, error) {
|
||||||
|
stale := time.Now().Add(-modelRecheckGap)
|
||||||
|
var rows []model.AiModelCache
|
||||||
|
err := s.db.WithContext(ctx).
|
||||||
|
Where("channel_id = ? AND capability IN ?", channelID, []string{"CHAT", ""}).
|
||||||
|
Where("checked_at IS NULL OR (unusable = ? AND checked_at < ?)", true, stale).
|
||||||
|
Order("id ASC").Limit(validateBatchCap).Find(&rows).Error
|
||||||
|
return rows, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateChannelModels 逐个试调渠道内待验证模型并落结论,把不可按需调用的模型
|
||||||
|
// 从池中剔除、把恢复供给的解除标记;ListModels 无字段可事先判别,只能试调习得。
|
||||||
|
func (s *AiGatewayService) validateChannelModels(ctx context.Context, channelID uint) {
|
||||||
|
if !s.beginValidate(channelID) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer s.endValidate(channelID)
|
||||||
|
var ch model.AiChannel
|
||||||
|
if err := s.db.WithContext(ctx).First(&ch, channelID).Error; err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
cred, err := s.configs.credentialsByID(ctx, ch.OciConfigID)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
rows, err := s.validateTargets(ctx, channelID)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("validate channel %d models: %v", channelID, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
for _, r := range rows {
|
||||||
|
s.validateOne(ctx, cred, &ch, r)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateOne 试调单个模型并落库:可用(200/429)清除标记;模型级不可用标记剔除;
|
||||||
|
// 其余 4xx 记录已检不改状态;5xx/网络类瞬态不落 checked_at,下轮再验。
|
||||||
|
func (s *AiGatewayService) validateOne(ctx context.Context, cred oci.Credentials, ch *model.AiChannel, r model.AiModelCache) {
|
||||||
|
code, err := s.client.GenAiProbeChat(ctx, cred, ch.Region, r.ModelOcid, r.Name)
|
||||||
|
updates := map[string]any{"checked_at": time.Now()}
|
||||||
|
switch {
|
||||||
|
case code == 200 || code == 429:
|
||||||
|
updates["unusable"], updates["unusable_reason"] = false, ""
|
||||||
|
case oci.IsModelUnavailable(err):
|
||||||
|
updates["unusable"], updates["unusable_reason"] = true, shortReason(oci.CompactError(err))
|
||||||
|
case code == 0 || code >= 500:
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.db.WithContext(ctx).Model(&model.AiModelCache{}).Where("id = ?", r.ID).Updates(updates)
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateModelsAsync 后台验证渠道模型池(手动同步后触发,不阻塞接口响应)。
|
||||||
|
func (s *AiGatewayService) validateModelsAsync(ctx context.Context, channelID uint) {
|
||||||
|
s.wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer s.wg.Done()
|
||||||
|
vctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Minute)
|
||||||
|
defer cancel()
|
||||||
|
s.validateChannelModels(vctx, channelID)
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
// probeCandidateCap 是单次探测的候选上限:模型级不可用会被标记剔除不再重试,
|
||||||
|
// 放宽到 8 让一次探测有机会越过整批坏模型找到可用者;其他错误另有 3 次止损预算。
|
||||||
|
const probeCandidateCap = 8
|
||||||
|
|
||||||
|
// probeCandidates 只取对话模型,按可靠度排序后跨厂商取候选:主流文本模型优先;
|
||||||
|
// voice 等负分形态(元数据标 CHAT 但实际不可对话)直接排除,不浪费试调预算;
|
||||||
|
// 每厂商先取最高分再按分数补位——部分区域某厂商全为微调基座(调用必失败),
|
||||||
|
// 不能让单一厂商占满候选名额拖垮整个渠道的探测结论。
|
||||||
func probeCandidates(models []oci.GenAiModel) []oci.GenAiModel {
|
func probeCandidates(models []oci.GenAiModel) []oci.GenAiModel {
|
||||||
var sorted []oci.GenAiModel
|
var sorted []oci.GenAiModel
|
||||||
for _, m := range models {
|
for _, m := range models {
|
||||||
if m.Capability == "" || m.Capability == "CHAT" {
|
if (m.Capability == "" || m.Capability == "CHAT") && probeScore(m.Name) >= 0 {
|
||||||
sorted = append(sorted, m)
|
sorted = append(sorted, m)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
sort.SliceStable(sorted, func(i, j int) bool {
|
sort.SliceStable(sorted, func(i, j int) bool {
|
||||||
return probeScore(sorted[i].Name) > probeScore(sorted[j].Name)
|
return probeScore(sorted[i].Name) > probeScore(sorted[j].Name)
|
||||||
})
|
})
|
||||||
if len(sorted) > 3 {
|
return diversifyByVendor(sorted, probeCandidateCap)
|
||||||
sorted = sorted[:3]
|
}
|
||||||
|
|
||||||
|
// diversifyByVendor 从已排序列表先每厂商各取一个,不足 limit 再按原序补位。
|
||||||
|
func diversifyByVendor(sorted []oci.GenAiModel, limit int) []oci.GenAiModel {
|
||||||
|
picked := make([]oci.GenAiModel, 0, limit)
|
||||||
|
used := make(map[int]bool)
|
||||||
|
seenVendor := make(map[string]bool)
|
||||||
|
for i, m := range sorted {
|
||||||
|
if len(picked) >= limit {
|
||||||
|
break
|
||||||
}
|
}
|
||||||
return sorted
|
if v := modelVendor(m); !seenVendor[v] {
|
||||||
|
seenVendor[v] = true
|
||||||
|
used[i] = true
|
||||||
|
picked = append(picked, m)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for i, m := range sorted {
|
||||||
|
if len(picked) >= limit {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
if !used[i] {
|
||||||
|
picked = append(picked, m)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return picked
|
||||||
|
}
|
||||||
|
|
||||||
|
// modelVendor 取厂商标识;OCI 未回填 vendor 时退化为模型名「.」前缀。
|
||||||
|
func modelVendor(m oci.GenAiModel) string {
|
||||||
|
if m.Vendor != "" {
|
||||||
|
return strings.ToLower(m.Vendor)
|
||||||
|
}
|
||||||
|
name := strings.ToLower(m.Name)
|
||||||
|
if i := strings.Index(name, "."); i > 0 {
|
||||||
|
return name[:i]
|
||||||
|
}
|
||||||
|
return name
|
||||||
}
|
}
|
||||||
|
|
||||||
func probeScore(name string) int {
|
func probeScore(name string) int {
|
||||||
@@ -377,7 +551,8 @@ func truncateErr(msg string) string {
|
|||||||
return msg
|
return msg
|
||||||
}
|
}
|
||||||
|
|
||||||
// SyncModels 重新拉取渠道区域的模型列表并覆盖缓存。
|
// SyncModels 重新拉取渠道区域的模型列表并覆盖缓存(标记按 OCID 结转),
|
||||||
|
// 随后触发后台验证:新模型逐个试调,不可按需调用的数十秒内从池中剔除。
|
||||||
func (s *AiGatewayService) SyncModels(ctx context.Context, id uint) ([]model.AiModelCache, error) {
|
func (s *AiGatewayService) SyncModels(ctx context.Context, id uint) ([]model.AiModelCache, error) {
|
||||||
var ch model.AiChannel
|
var ch model.AiChannel
|
||||||
if err := s.db.WithContext(ctx).First(&ch, id).Error; err != nil {
|
if err := s.db.WithContext(ctx).First(&ch, id).Error; err != nil {
|
||||||
@@ -394,16 +569,26 @@ func (s *AiGatewayService) SyncModels(ctx context.Context, id uint) ([]model.AiM
|
|||||||
if err := s.replaceModels(ctx, id, models); err != nil {
|
if err := s.replaceModels(ctx, id, models); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
s.validateModelsAsync(ctx, id)
|
||||||
return s.channelModels(ctx, id)
|
return s.channelModels(ctx, id)
|
||||||
}
|
}
|
||||||
|
|
||||||
// replaceModels 以事务整组覆盖渠道模型缓存。
|
// replaceModels 以事务整组覆盖渠道模型缓存,按 OCID 结转不可用标记与验证时间
|
||||||
|
// (OCID 变化视为新条目,自然回到待验证状态)。
|
||||||
func (s *AiGatewayService) replaceModels(ctx context.Context, channelID uint, models []oci.GenAiModel) error {
|
func (s *AiGatewayService) replaceModels(ctx context.Context, channelID uint, models []oci.GenAiModel) error {
|
||||||
|
marks, err := s.loadModelMarks(ctx, channelID)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
rows := make([]model.AiModelCache, 0, len(models))
|
rows := make([]model.AiModelCache, 0, len(models))
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
for _, m := range models {
|
for _, m := range models {
|
||||||
rows = append(rows, model.AiModelCache{ChannelID: channelID, ModelOcid: m.Ocid, Name: m.Name, Vendor: m.Vendor,
|
row := model.AiModelCache{ChannelID: channelID, ModelOcid: m.Ocid, Name: m.Name, Vendor: m.Vendor,
|
||||||
Capability: m.Capability, SyncedAt: now, DeprecatedAt: m.Deprecated, RetiredAt: m.Retired})
|
Capability: m.Capability, SyncedAt: now, DeprecatedAt: m.Deprecated, RetiredAt: m.Retired}
|
||||||
|
if prev, ok := marks[m.Ocid]; ok {
|
||||||
|
row.Unusable, row.UnusableReason, row.CheckedAt = prev.Unusable, prev.UnusableReason, prev.CheckedAt
|
||||||
|
}
|
||||||
|
rows = append(rows, row)
|
||||||
}
|
}
|
||||||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||||
if err := tx.Where("channel_id = ?", channelID).Delete(&model.AiModelCache{}).Error; err != nil {
|
if err := tx.Where("channel_id = ?", channelID).Delete(&model.AiModelCache{}).Error; err != nil {
|
||||||
@@ -416,17 +601,33 @@ func (s *AiGatewayService) replaceModels(ctx context.Context, channelID uint, mo
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// loadModelMarks 取渠道内带标记或已验证过的行(OCID → 行),供同步结转与候选过滤。
|
||||||
|
func (s *AiGatewayService) loadModelMarks(ctx context.Context, channelID uint) (map[string]model.AiModelCache, error) {
|
||||||
|
var rows []model.AiModelCache
|
||||||
|
err := s.db.WithContext(ctx).Select("model_ocid", "unusable", "unusable_reason", "checked_at").
|
||||||
|
Where("channel_id = ? AND (unusable = ? OR checked_at IS NOT NULL)", channelID, true).Find(&rows).Error
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("load model marks: %w", err)
|
||||||
|
}
|
||||||
|
marks := make(map[string]model.AiModelCache, len(rows))
|
||||||
|
for _, r := range rows {
|
||||||
|
marks[r.ModelOcid] = r
|
||||||
|
}
|
||||||
|
return marks, nil
|
||||||
|
}
|
||||||
|
|
||||||
func (s *AiGatewayService) channelModels(ctx context.Context, channelID uint) ([]model.AiModelCache, error) {
|
func (s *AiGatewayService) channelModels(ctx context.Context, channelID uint) ([]model.AiModelCache, error) {
|
||||||
var rows []model.AiModelCache
|
var rows []model.AiModelCache
|
||||||
err := s.db.WithContext(ctx).Where("channel_id = ?", channelID).Order("name ASC").Find(&rows).Error
|
err := s.db.WithContext(ctx).Where("channel_id = ?", channelID).Order("name ASC").Find(&rows).Error
|
||||||
return rows, err
|
return rows, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// GatewayModels 聚合启用渠道的模型(按名称去重),供 /ai/v1/models;
|
// GatewayModels 聚合启用渠道的可用模型(按名称去重,剔除不可按需调用标记),
|
||||||
// group 非空时仅聚合该分组渠道(与密钥分组路由口径一致)。
|
// 供 /ai/v1/models;group 非空时仅聚合该分组渠道(与密钥分组路由口径一致)。
|
||||||
func (s *AiGatewayService) GatewayModels(ctx context.Context, group string) (aiwire.ModelList, error) {
|
func (s *AiGatewayService) GatewayModels(ctx context.Context, group string) (aiwire.ModelList, error) {
|
||||||
q := s.db.WithContext(ctx).
|
q := s.db.WithContext(ctx).
|
||||||
Joins("JOIN ai_channels ON ai_channels.id = ai_model_caches.channel_id AND ai_channels.enabled = ?", true)
|
Joins("JOIN ai_channels ON ai_channels.id = ai_model_caches.channel_id AND ai_channels.enabled = ?", true).
|
||||||
|
Where("(ai_model_caches.unusable = ? OR ai_model_caches.unusable IS NULL)", false)
|
||||||
if group != "" {
|
if group != "" {
|
||||||
q = q.Where("ai_channels.channel_group = ?", group)
|
q = q.Where("ai_channels.channel_group = ?", group)
|
||||||
}
|
}
|
||||||
@@ -457,6 +658,7 @@ func (s *AiGatewayService) DeprecatingModels(ctx context.Context, within time.Du
|
|||||||
err := s.db.WithContext(ctx).
|
err := s.db.WithContext(ctx).
|
||||||
Where("(retired_at IS NOT NULL AND retired_at > ? AND retired_at <= ?) OR (deprecated_at IS NOT NULL AND deprecated_at >= ? AND deprecated_at <= ?)",
|
Where("(retired_at IS NOT NULL AND retired_at > ? AND retired_at <= ?) OR (deprecated_at IS NOT NULL AND deprecated_at >= ? AND deprecated_at <= ?)",
|
||||||
now, deadline, now, deadline).
|
now, deadline, now, deadline).
|
||||||
|
Where("(unusable = ? OR unusable IS NULL)", false).
|
||||||
Order("name ASC").Find(&rows).Error
|
Order("name ASC").Find(&rows).Error
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -477,7 +679,7 @@ func (s *AiGatewayService) DeprecatingModels(ctx context.Context, within time.Du
|
|||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// ProbeAll 逐个探测全部渠道,返回状态汇总;供 AI 探测后台任务调用。
|
// ProbeAll 逐个探测全部渠道并顺带验证模型池,返回状态汇总;供 AI 探测后台任务调用。
|
||||||
func (s *AiGatewayService) ProbeAll(ctx context.Context) (string, error) {
|
func (s *AiGatewayService) ProbeAll(ctx context.Context) (string, error) {
|
||||||
chs, err := s.Channels(ctx)
|
chs, err := s.Channels(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -493,6 +695,8 @@ func (s *AiGatewayService) ProbeAll(ctx context.Context) (string, error) {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
counts[fresh.ProbeStatus]++
|
counts[fresh.ProbeStatus]++
|
||||||
|
// 后台任务里同步执行:新模型验证入池、坏模型定期复检,零人工收敛
|
||||||
|
s.validateChannelModels(ctx, ch.ID)
|
||||||
}
|
}
|
||||||
msg := fmt.Sprintf("probed %d: %d ok, %d no_service, %d no_quota, %d error",
|
msg := fmt.Sprintf("probed %d: %d ok, %d no_service, %d no_quota, %d error",
|
||||||
len(chs), counts["ok"], counts["no_service"], counts["no_quota"], counts["error"])
|
len(chs), counts["ok"], counts["no_service"], counts["no_quota"], counts["error"])
|
||||||
|
|||||||
@@ -43,10 +43,13 @@ func (s *AiGatewayService) Chat(ctx context.Context, ir aiwire.ChatRequest, grou
|
|||||||
s.markSuccess(ctx, cand.ch.ID)
|
s.markSuccess(ctx, cand.ch.ID)
|
||||||
return resp, meta, nil
|
return resp, meta, nil
|
||||||
}
|
}
|
||||||
if !retryable(err) {
|
retry, penalize := s.noteCallErr(ctx, cand, err)
|
||||||
|
if !retry {
|
||||||
return nil, meta, err
|
return nil, meta, err
|
||||||
}
|
}
|
||||||
|
if penalize {
|
||||||
s.markFailure(ctx, cand.ch.ID)
|
s.markFailure(ctx, cand.ch.ID)
|
||||||
|
}
|
||||||
excluded[cand.ch.ID] = true
|
excluded[cand.ch.ID] = true
|
||||||
meta.Retries++
|
meta.Retries++
|
||||||
lastErr = err
|
lastErr = err
|
||||||
@@ -83,10 +86,13 @@ func (s *AiGatewayService) OpenStream(ctx context.Context, ir aiwire.ChatRequest
|
|||||||
s.markSuccess(ctx, cand.ch.ID)
|
s.markSuccess(ctx, cand.ch.ID)
|
||||||
return stream, meta, nil
|
return stream, meta, nil
|
||||||
}
|
}
|
||||||
if !retryable(err) {
|
retry, penalize := s.noteCallErr(ctx, cand, err)
|
||||||
|
if !retry {
|
||||||
return nil, meta, err
|
return nil, meta, err
|
||||||
}
|
}
|
||||||
|
if penalize {
|
||||||
s.markFailure(ctx, cand.ch.ID)
|
s.markFailure(ctx, cand.ch.ID)
|
||||||
|
}
|
||||||
excluded[cand.ch.ID] = true
|
excluded[cand.ch.ID] = true
|
||||||
meta.Retries++
|
meta.Retries++
|
||||||
lastErr = err
|
lastErr = err
|
||||||
@@ -102,12 +108,18 @@ func firstErr(lastErr, pickErr error) error {
|
|||||||
return pickErr
|
return pickErr
|
||||||
}
|
}
|
||||||
|
|
||||||
// retryable 判定是否换渠道重试:429 / 5xx / 网络错误可重试,其余 4xx 直接透传。
|
// noteCallErr 汇总一次调用失败:模型级不可用(微调基座 400 / 实体不存在 404)先标记
|
||||||
func retryable(err error) bool {
|
// 剔除该 (渠道, 模型),换渠道重试且不计熔断——这是模型×区域供给问题而非渠道健康问题;
|
||||||
if status, ok := oci.ServiceStatus(err); ok {
|
// 429 / 5xx / 网络错误换渠道并计熔断;其余 4xx 直接透传。
|
||||||
return status == 429 || status >= 500
|
func (s *AiGatewayService) noteCallErr(ctx context.Context, cand *aiCandidate, err error) (retry, penalize bool) {
|
||||||
|
if oci.IsModelUnavailable(err) {
|
||||||
|
s.markModelUnusable(ctx, cand.ch.ID, cand.modelOcid, oci.CompactError(err))
|
||||||
|
return true, false
|
||||||
}
|
}
|
||||||
return true
|
if status, ok := oci.ServiceStatus(err); ok {
|
||||||
|
return status == 429 || status >= 500, true
|
||||||
|
}
|
||||||
|
return true, true
|
||||||
}
|
}
|
||||||
|
|
||||||
// pick 选出支持该模型的最优渠道:能力匹配 → 启用 → 分组匹配 → 未熔断 → 最小优先级组 → 加权随机。
|
// pick 选出支持该模型的最优渠道:能力匹配 → 启用 → 分组匹配 → 未熔断 → 最小优先级组 → 加权随机。
|
||||||
@@ -142,10 +154,11 @@ func (s *AiGatewayService) pick(ctx context.Context, modelName, group, capabilit
|
|||||||
return &aiCandidate{ch: chosen, modelOcid: ocids[chosen.ID]}, nil
|
return &aiCandidate{ch: chosen, modelOcid: ocids[chosen.ID]}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// modelChannels 查出提供该模型的渠道 ID 及各自的模型 OCID;
|
// modelChannels 查出提供该模型的渠道 ID 及各自的模型 OCID(剔除不可按需调用标记);
|
||||||
// capability=CHAT 时兼容存量空串(加列前只同步对话模型)。
|
// capability=CHAT 时兼容存量空串(加列前只同步对话模型);unusable 需兼容 NULL
|
||||||
|
// (AutoMigrate 加列后、首次重同步前的存量行)。
|
||||||
func (s *AiGatewayService) modelChannels(ctx context.Context, modelName, capability string) (map[uint]string, []uint, error) {
|
func (s *AiGatewayService) modelChannels(ctx context.Context, modelName, capability string) (map[uint]string, []uint, error) {
|
||||||
q := s.db.WithContext(ctx).Where("name = ?", modelName)
|
q := s.db.WithContext(ctx).Where("name = ? AND (unusable = ? OR unusable IS NULL)", modelName, false)
|
||||||
if capability == "CHAT" {
|
if capability == "CHAT" {
|
||||||
q = q.Where("capability IN ?", []string{"CHAT", ""})
|
q = q.Where("capability IN ?", []string{"CHAT", ""})
|
||||||
} else {
|
} else {
|
||||||
@@ -221,10 +234,13 @@ func (s *AiGatewayService) Embeddings(ctx context.Context, req aiwire.Embeddings
|
|||||||
s.markSuccess(ctx, cand.ch.ID)
|
s.markSuccess(ctx, cand.ch.ID)
|
||||||
return resp, meta, nil
|
return resp, meta, nil
|
||||||
}
|
}
|
||||||
if !retryable(err) {
|
retry, penalize := s.noteCallErr(ctx, cand, err)
|
||||||
|
if !retry {
|
||||||
return nil, meta, err
|
return nil, meta, err
|
||||||
}
|
}
|
||||||
|
if penalize {
|
||||||
s.markFailure(ctx, cand.ch.ID)
|
s.markFailure(ctx, cand.ch.ID)
|
||||||
|
}
|
||||||
excluded[cand.ch.ID] = true
|
excluded[cand.ch.ID] = true
|
||||||
meta.Retries++
|
meta.Retries++
|
||||||
lastErr = err
|
lastErr = err
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package service
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -13,14 +14,28 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// stubServiceError 实现 common.ServiceError,用于模拟带状态码的 OCI 服务端错误。
|
// stubServiceError 实现 common.ServiceError,用于模拟带状态码的 OCI 服务端错误。
|
||||||
type stubServiceError struct{ status int }
|
type stubServiceError struct {
|
||||||
|
status int
|
||||||
|
msg string
|
||||||
|
}
|
||||||
|
|
||||||
func (e stubServiceError) Error() string { return "stub service error" }
|
func (e stubServiceError) Error() string { return "stub service error" }
|
||||||
func (e stubServiceError) GetHTTPStatusCode() int { return e.status }
|
func (e stubServiceError) GetHTTPStatusCode() int { return e.status }
|
||||||
func (e stubServiceError) GetMessage() string { return "stub" }
|
func (e stubServiceError) GetMessage() string {
|
||||||
|
if e.msg != "" {
|
||||||
|
return e.msg
|
||||||
|
}
|
||||||
|
return "stub"
|
||||||
|
}
|
||||||
func (e stubServiceError) GetCode() string { return "Stub" }
|
func (e stubServiceError) GetCode() string { return "Stub" }
|
||||||
func (e stubServiceError) GetOpcRequestID() string { return "req-1" }
|
func (e stubServiceError) GetOpcRequestID() string { return "req-1" }
|
||||||
|
|
||||||
|
// finetuneBaseErr 模拟「微调基座模型不可按需调用」的 OCI 400。
|
||||||
|
func finetuneBaseErr() stubServiceError {
|
||||||
|
return stubServiceError{status: 400,
|
||||||
|
msg: "Not allowed to call finetune base model ocid1.generativeaimodel.oc1.eu-frankfurt-1.tpel5q, use Endpoint: false"}
|
||||||
|
}
|
||||||
|
|
||||||
// gatewayStubClient 覆写 GenAI 四方法;chatErrs 逐次弹出以模拟先失败后成功。
|
// gatewayStubClient 覆写 GenAI 四方法;chatErrs 逐次弹出以模拟先失败后成功。
|
||||||
type gatewayStubClient struct {
|
type gatewayStubClient struct {
|
||||||
*fakeClient
|
*fakeClient
|
||||||
@@ -29,6 +44,8 @@ type gatewayStubClient struct {
|
|||||||
modelsErr error
|
modelsErr error
|
||||||
probeCode int
|
probeCode int
|
||||||
probeErr error
|
probeErr error
|
||||||
|
// probeSeq 非空时逐次弹出,模拟按候选依次试调;弹尽后回落 probeCode/probeErr
|
||||||
|
probeSeq []probeResult
|
||||||
chatResp *aiwire.ChatResponse
|
chatResp *aiwire.ChatResponse
|
||||||
chatErrs []error
|
chatErrs []error
|
||||||
chatCalls int
|
chatCalls int
|
||||||
@@ -47,7 +64,18 @@ func (f *gatewayStubClient) ListGenAiModels(ctx context.Context, cred oci.Creden
|
|||||||
return f.models, f.modelsErr
|
return f.models, f.modelsErr
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// probeResult 是 gatewayStubClient.probeSeq 的单次探测结果。
|
||||||
|
type probeResult struct {
|
||||||
|
code int
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
func (f *gatewayStubClient) GenAiProbeChat(ctx context.Context, cred oci.Credentials, region, modelOcid, modelName string) (int, error) {
|
func (f *gatewayStubClient) GenAiProbeChat(ctx context.Context, cred oci.Credentials, region, modelOcid, modelName string) (int, error) {
|
||||||
|
if len(f.probeSeq) > 0 {
|
||||||
|
r := f.probeSeq[0]
|
||||||
|
f.probeSeq = f.probeSeq[1:]
|
||||||
|
return r.code, r.err
|
||||||
|
}
|
||||||
return f.probeCode, f.probeErr
|
return f.probeCode, f.probeErr
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -400,6 +428,263 @@ func TestProbeCandidates(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestProbeCandidatesVendorDiversity(t *testing.T) {
|
||||||
|
// 部分区域单一厂商全为微调基座:候选须跨厂商分散,不能被 3 个 llama 占满前排
|
||||||
|
models := []oci.GenAiModel{
|
||||||
|
{Ocid: "l1", Name: "meta.llama-3-70b-instruct"},
|
||||||
|
{Ocid: "l2", Name: "meta.llama-3.1-405b-instruct"},
|
||||||
|
{Ocid: "l3", Name: "meta.llama-3.3-70b-instruct"},
|
||||||
|
{Ocid: "c1", Name: "cohere.command-a-03-2025"},
|
||||||
|
{Ocid: "g1", Name: "xai.grok-4"},
|
||||||
|
}
|
||||||
|
got := probeCandidates(models)
|
||||||
|
if len(got) != 5 || got[0].Name != "meta.llama-3-70b-instruct" {
|
||||||
|
t.Fatalf("上限内全量返回且最高分居首: %+v", got)
|
||||||
|
}
|
||||||
|
vendors := map[string]bool{}
|
||||||
|
for _, m := range got[:3] {
|
||||||
|
vendors[modelVendor(m)] = true
|
||||||
|
}
|
||||||
|
if len(vendors) != 3 {
|
||||||
|
t.Errorf("前 3 个候选应覆盖 3 个厂商: %+v", got)
|
||||||
|
}
|
||||||
|
// 超过上限时截断到 probeCandidateCap
|
||||||
|
var many []oci.GenAiModel
|
||||||
|
for i := 0; i < 12; i++ {
|
||||||
|
many = append(many, oci.GenAiModel{Ocid: fmt.Sprintf("m%d", i), Name: fmt.Sprintf("meta.llama-%d", i)})
|
||||||
|
}
|
||||||
|
if capped := probeCandidates(many); len(capped) != probeCandidateCap {
|
||||||
|
t.Errorf("候选应截断到 %d: got %d", probeCandidateCap, len(capped))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// entityNotFoundErr 模拟「实体不存在」404(模型在区域内无按需供给)。
|
||||||
|
func entityNotFoundErr() stubServiceError {
|
||||||
|
return stubServiceError{status: 404,
|
||||||
|
msg: "Entity with key ocid1.generativeaimodel.oc1.eu-frankfurt-1.2flsfq not found"}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestProbeSkipsUnavailableModels(t *testing.T) {
|
||||||
|
// 首选 llama 不可按需调用(基座 400 / 实体 404)→ 标记剔除 → 换候选成功 → 渠道判可用
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
bad probeResult
|
||||||
|
}{
|
||||||
|
{"微调基座 400", probeResult{400, finetuneBaseErr()}},
|
||||||
|
{"实体不存在 404", probeResult{404, entityNotFoundErr()}},
|
||||||
|
}
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
client := &gatewayStubClient{
|
||||||
|
fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}},
|
||||||
|
models: []oci.GenAiModel{
|
||||||
|
{Ocid: "m1", Name: "meta.llama-3-70b-instruct", Vendor: "meta"},
|
||||||
|
{Ocid: "m2", Name: "meta.llama-3.1-70b-instruct", Vendor: "meta"},
|
||||||
|
{Ocid: "m3", Name: "cohere.command-a-03-2025", Vendor: "cohere"},
|
||||||
|
},
|
||||||
|
probeSeq: []probeResult{tt.bad, {200, nil}},
|
||||||
|
}
|
||||||
|
gw, svc := newTestGateway(t, client)
|
||||||
|
cfg := importAliveConfig(t, svc)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
ch, err := gw.CreateChannel(ctx, ChannelInput{OciConfigID: cfg.ID, Region: "eu-frankfurt-1"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("CreateChannel: %v", err)
|
||||||
|
}
|
||||||
|
probed, err := gw.ProbeChannel(ctx, ch.ID)
|
||||||
|
if err != nil || probed.ProbeStatus != "ok" {
|
||||||
|
t.Fatalf("坏候选后应换候选并判可用: %+v, %v", probed, err)
|
||||||
|
}
|
||||||
|
var row model.AiModelCache
|
||||||
|
if err := gw.db.Where("channel_id = ? AND model_ocid = ?", ch.ID, "m1").First(&row).Error; err != nil || !row.Unusable {
|
||||||
|
t.Fatalf("首个坏候选应被标记不可用: %+v, %v", row, err)
|
||||||
|
}
|
||||||
|
// 再次探测:同步保留标记,m1 不再进候选(probeSeq 只需一次 200)
|
||||||
|
client.probeSeq = []probeResult{{200, nil}}
|
||||||
|
probed, err = gw.ProbeChannel(ctx, ch.ID)
|
||||||
|
if err != nil || probed.ProbeStatus != "ok" {
|
||||||
|
t.Fatalf("复测应跳过已标记模型: %+v, %v", probed, err)
|
||||||
|
}
|
||||||
|
var again model.AiModelCache
|
||||||
|
gw.db.Where("channel_id = ? AND model_ocid = ?", ch.ID, "m1").First(&again)
|
||||||
|
if !again.Unusable {
|
||||||
|
t.Error("探测触发的同步应保留不可用标记")
|
||||||
|
}
|
||||||
|
// 手动同步同样保留标记(坏模型不随重新同步复活);近期已检的标记不被后台验证翻转
|
||||||
|
client.probeCode = 200
|
||||||
|
if _, err := gw.SyncModels(ctx, ch.ID); err != nil {
|
||||||
|
t.Fatalf("SyncModels: %v", err)
|
||||||
|
}
|
||||||
|
gw.Wait()
|
||||||
|
var kept model.AiModelCache
|
||||||
|
gw.db.Where("channel_id = ? AND model_ocid = ?", ch.ID, "m1").First(&kept)
|
||||||
|
if !kept.Unusable {
|
||||||
|
t.Error("手动同步不应清除不可用标记")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestValidateModelsAfterSync(t *testing.T) {
|
||||||
|
// 同步后后台验证:坏模型标记剔除、好模型记录已检、其他 4xx 不改状态、非对话模型不试调
|
||||||
|
client := &gatewayStubClient{
|
||||||
|
fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}},
|
||||||
|
models: []oci.GenAiModel{
|
||||||
|
{Ocid: "v1", Name: "meta.llama-4-maverick", Vendor: "meta", Capability: "CHAT"},
|
||||||
|
{Ocid: "v2", Name: "xai.grok-4", Vendor: "xai", Capability: "CHAT"},
|
||||||
|
{Ocid: "v3", Name: "xai.grok-voice-agent", Vendor: "xai", Capability: "CHAT"},
|
||||||
|
{Ocid: "v4", Name: "cohere.embed-v4.0", Vendor: "cohere", Capability: "EMBEDDING"},
|
||||||
|
},
|
||||||
|
probeSeq: []probeResult{{404, entityNotFoundErr()}, {200, nil}, {400, stubServiceError{status: 400}}},
|
||||||
|
}
|
||||||
|
gw, svc := newTestGateway(t, client)
|
||||||
|
cfg := importAliveConfig(t, svc)
|
||||||
|
ctx := context.Background()
|
||||||
|
ch, err := gw.CreateChannel(ctx, ChannelInput{OciConfigID: cfg.ID, Region: "us-ashburn-1"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("CreateChannel: %v", err)
|
||||||
|
}
|
||||||
|
if _, err := gw.SyncModels(ctx, ch.ID); err != nil {
|
||||||
|
t.Fatalf("SyncModels: %v", err)
|
||||||
|
}
|
||||||
|
gw.Wait()
|
||||||
|
|
||||||
|
want := map[string]struct {
|
||||||
|
unusable bool
|
||||||
|
checked bool
|
||||||
|
}{
|
||||||
|
"v1": {true, true}, // 实体 404 → 标记剔除
|
||||||
|
"v2": {false, true}, // 200 → 可用已检
|
||||||
|
"v3": {false, true}, // 普通 400 → 已检不标记
|
||||||
|
"v4": {false, false}, // EMBEDDING 不试调
|
||||||
|
}
|
||||||
|
rows, _ := gw.channelModels(ctx, ch.ID)
|
||||||
|
for _, r := range rows {
|
||||||
|
w := want[r.ModelOcid]
|
||||||
|
if r.Unusable != w.unusable || (r.CheckedAt != nil) != w.checked {
|
||||||
|
t.Errorf("%s: unusable=%v checked=%v, want %+v", r.ModelOcid, r.Unusable, r.CheckedAt != nil, w)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
list, err := gw.GatewayModels(ctx, "")
|
||||||
|
if err != nil || len(list.Data) != 3 {
|
||||||
|
t.Errorf("网关列表应只剔除被标记的坏模型(余 v2/v3/v4), got %+v, %v", list.Data, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestValidateRecheckUnmarksRecovered(t *testing.T) {
|
||||||
|
// 已标记模型超过复检间隔后重验:恢复供给(200)自动解除标记
|
||||||
|
client := &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}, probeCode: 200}
|
||||||
|
gw, svc := newTestGateway(t, client)
|
||||||
|
cfg := importAliveConfig(t, svc)
|
||||||
|
ch := seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1)
|
||||||
|
ctx := context.Background()
|
||||||
|
old := time.Now().Add(-25 * time.Hour)
|
||||||
|
gw.db.Model(&model.AiModelCache{}).Where("channel_id = ?", ch.ID).
|
||||||
|
Updates(map[string]any{"unusable": true, "unusable_reason": "x", "checked_at": old})
|
||||||
|
|
||||||
|
gw.validateChannelModels(ctx, ch.ID)
|
||||||
|
var row model.AiModelCache
|
||||||
|
gw.db.Where("channel_id = ?", ch.ID).First(&row)
|
||||||
|
if row.Unusable || row.UnusableReason != "" || row.CheckedAt == nil || !row.CheckedAt.After(old) {
|
||||||
|
t.Errorf("超期复检应解除标记并刷新已检时间: %+v", row)
|
||||||
|
}
|
||||||
|
// 未超期的标记不复检(probeSeq 为空、fallback 200 也不会被消费)
|
||||||
|
fresh := time.Now()
|
||||||
|
gw.db.Model(&model.AiModelCache{}).Where("channel_id = ?", ch.ID).
|
||||||
|
Updates(map[string]any{"unusable": true, "checked_at": fresh})
|
||||||
|
gw.validateChannelModels(ctx, ch.ID)
|
||||||
|
gw.db.Where("channel_id = ?", ch.ID).First(&row)
|
||||||
|
if !row.Unusable {
|
||||||
|
t.Error("未超期的标记不应被复检翻转")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestProbeAuth404StillNoQuota(t *testing.T) {
|
||||||
|
// 鉴权类 404(NotAuthorizedOrNotFound)仍属租户级,直接定论 no_quota 且不标记模型
|
||||||
|
client := &gatewayStubClient{
|
||||||
|
fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}},
|
||||||
|
models: []oci.GenAiModel{{Ocid: "m1", Name: "meta.llama-3.3-70b-instruct", Vendor: "meta"}},
|
||||||
|
probeCode: 404,
|
||||||
|
probeErr: stubServiceError{status: 404, msg: "Authorization failed or requested resource not found."},
|
||||||
|
}
|
||||||
|
gw, svc := newTestGateway(t, client)
|
||||||
|
cfg := importAliveConfig(t, svc)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
ch, _ := gw.CreateChannel(ctx, ChannelInput{OciConfigID: cfg.ID, Region: "eu-frankfurt-1"})
|
||||||
|
probed, _ := gw.ProbeChannel(ctx, ch.ID)
|
||||||
|
if probed.ProbeStatus != "no_quota" {
|
||||||
|
t.Errorf("鉴权 404 status = %q, want no_quota", probed.ProbeStatus)
|
||||||
|
}
|
||||||
|
var marked int64
|
||||||
|
gw.db.Model(&model.AiModelCache{}).Where("unusable = ?", true).Count(&marked)
|
||||||
|
if marked != 0 {
|
||||||
|
t.Errorf("鉴权 404 不应标记模型, marked=%d", marked)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAiChatFinetuneSwitchesChannelWithoutPenalty(t *testing.T) {
|
||||||
|
// 微调基座 400 换渠道重试成功,且不计入熔断失败
|
||||||
|
client := &gatewayStubClient{
|
||||||
|
fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}},
|
||||||
|
chatResp: &aiwire.ChatResponse{Model: "meta.llama-3.3-70b-instruct", Choices: []aiwire.Choice{{Message: aiwire.ChatMessage{Role: "assistant", Content: aiwire.NewTextContent("hi")}, FinishReason: "stop"}}},
|
||||||
|
chatErrs: []error{finetuneBaseErr()},
|
||||||
|
}
|
||||||
|
gw, svc := newTestGateway(t, client)
|
||||||
|
cfg := importAliveConfig(t, svc)
|
||||||
|
seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1)
|
||||||
|
seedChannel(t, gw, cfg.ID, "us-chicago-1", 1, 1)
|
||||||
|
|
||||||
|
resp, meta, err := gw.Chat(context.Background(), aiwire.ChatRequest{Model: "meta.llama-3.3-70b-instruct", Messages: []aiwire.ChatMessage{{Role: "user", Content: aiwire.NewTextContent("你好")}}}, "")
|
||||||
|
if err != nil || resp == nil {
|
||||||
|
t.Fatalf("Chat = %v, %v", resp, err)
|
||||||
|
}
|
||||||
|
if meta.Retries != 1 || client.chatCalls != 2 {
|
||||||
|
t.Errorf("应换渠道重试一次: retries=%d calls=%d", meta.Retries, client.chatCalls)
|
||||||
|
}
|
||||||
|
var chs []model.AiChannel
|
||||||
|
gw.db.Find(&chs)
|
||||||
|
for _, ch := range chs {
|
||||||
|
if ch.FailCount != 0 {
|
||||||
|
t.Errorf("微调基座 400 不应计入熔断: 渠道 %s failCount=%d", ch.Name, ch.FailCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// 失败渠道的该模型被标记,不再参与路由;成功渠道不受影响
|
||||||
|
region := client.regions[0]
|
||||||
|
var row model.AiModelCache
|
||||||
|
gw.db.Where("model_ocid = ?", "ocid1..m-"+region).First(&row)
|
||||||
|
if !row.Unusable {
|
||||||
|
t.Errorf("失败渠道的模型应被标记不可用: %+v", row)
|
||||||
|
}
|
||||||
|
var usable int64
|
||||||
|
gw.db.Model(&model.AiModelCache{}).Where("unusable = ?", false).Count(&usable)
|
||||||
|
if usable != 1 {
|
||||||
|
t.Errorf("成功渠道模型应保持可用, usable=%d", usable)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUnusableModelExcludedFromPoolAndRouting(t *testing.T) {
|
||||||
|
// 唯一渠道的模型被标记后:网关列表不再展示,路由按未知模型拒绝,渠道详情仍可见标记
|
||||||
|
gw, svc := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}})
|
||||||
|
cfg := importAliveConfig(t, svc)
|
||||||
|
ch := seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
gw.markModelUnusable(ctx, ch.ID, "ocid1..m-eu-frankfurt-1", "Entity with key … not found")
|
||||||
|
list, err := gw.GatewayModels(ctx, "")
|
||||||
|
if err != nil || len(list.Data) != 0 {
|
||||||
|
t.Errorf("已标记模型不应出现在网关列表: %+v, %v", list.Data, err)
|
||||||
|
}
|
||||||
|
if _, _, err := gw.Chat(ctx, aiwire.ChatRequest{Model: "meta.llama-3.3-70b-instruct"}, ""); !errors.Is(err, ErrAiUnknownModel) {
|
||||||
|
t.Errorf("已标记模型路由应拒绝: %v", err)
|
||||||
|
}
|
||||||
|
rows, err := gw.channelModels(ctx, ch.ID)
|
||||||
|
if err != nil || len(rows) != 1 || !rows[0].Unusable || rows[0].UnusableReason == "" {
|
||||||
|
t.Errorf("渠道详情应保留标记行便于排查: %+v, %v", rows, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestDeprecatingModels(t *testing.T) {
|
func TestDeprecatingModels(t *testing.T) {
|
||||||
gw, _ := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}})
|
gw, _ := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}})
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
@@ -507,3 +792,24 @@ func TestAiContentLogSwitch(t *testing.T) {
|
|||||||
t.Fatalf("关闭失败: %+v, %v", fresh.ContentLogUntil, err)
|
t.Fatalf("关闭失败: %+v, %v", fresh.ContentLogUntil, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestLegacyNullUnusableRowsStayVisible(t *testing.T) {
|
||||||
|
// 升级路径回归:AutoMigrate 加列后、首次重同步前,存量行 unusable 为 NULL,
|
||||||
|
// 网关列表与路由必须照常包含这些行,不能因 unusable = false 过滤而整池消失
|
||||||
|
gw, svc := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}})
|
||||||
|
cfg := importAliveConfig(t, svc)
|
||||||
|
ch := seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1)
|
||||||
|
ctx := context.Background()
|
||||||
|
if err := gw.db.Exec("UPDATE ai_model_caches SET unusable = NULL WHERE channel_id = ?", ch.ID).Error; err != nil {
|
||||||
|
t.Fatalf("set legacy null: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
list, err := gw.GatewayModels(ctx, "")
|
||||||
|
if err != nil || len(list.Data) != 1 {
|
||||||
|
t.Errorf("存量 NULL 行应仍在网关列表: %+v, %v", list.Data, err)
|
||||||
|
}
|
||||||
|
_, ids, err := gw.modelChannels(ctx, "meta.llama-3.3-70b-instruct", "CHAT")
|
||||||
|
if err != nil || len(ids) != 1 {
|
||||||
|
t.Errorf("存量 NULL 行应仍参与路由: %v, %v", ids, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -137,6 +137,13 @@ func newTestService(t *testing.T, client oci.Client) *OciConfigService {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("open in-memory sqlite: %v", err)
|
t.Fatalf("open in-memory sqlite: %v", err)
|
||||||
}
|
}
|
||||||
|
// :memory: 库每条连接各自独立,后台 goroutine(模型验证等)触发第二条连接会看到空库;
|
||||||
|
// 收敛到单连接,与 tenantdelete 测试环境一致
|
||||||
|
sqlDB, err := db.DB()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("database handle: %v", err)
|
||||||
|
}
|
||||||
|
sqlDB.SetMaxOpenConns(1)
|
||||||
if err := db.AutoMigrate(
|
if err := db.AutoMigrate(
|
||||||
&model.OciConfig{}, &model.RegionCache{}, &model.CompartmentCache{}, &model.Proxy{},
|
&model.OciConfig{}, &model.RegionCache{}, &model.CompartmentCache{}, &model.Proxy{},
|
||||||
); err != nil {
|
); err != nil {
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package service
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -46,6 +47,12 @@ type TaskService struct {
|
|||||||
entries map[uint]cron.EntryID
|
entries map[uint]cron.EntryID
|
||||||
// runMu 让租户删除等待在途执行结束,并阻止新执行读取删除前的任务快照。
|
// runMu 让租户删除等待在途执行结束,并阻止新执行读取删除前的任务快照。
|
||||||
runMu sync.RWMutex
|
runMu sync.RWMutex
|
||||||
|
// runningMu 保护 running:同一任务不允许并发重复执行
|
||||||
|
// (cron 重叠触发静默跳过,手动触发返回 ErrTaskRunning)
|
||||||
|
runningMu sync.Mutex
|
||||||
|
running map[uint]bool
|
||||||
|
// runWG 追踪手动触发的后台执行,Stop 时等待收尾
|
||||||
|
runWG sync.WaitGroup
|
||||||
|
|
||||||
// snatchVarsMu 保护 snatchVars:runSnatch 抢满时写入成功通知变量,
|
// snatchVarsMu 保护 snatchVars:runSnatch 抢满时写入成功通知变量,
|
||||||
// execute 组装执行后快照时取走(一次性)
|
// execute 组装执行后快照时取走(一次性)
|
||||||
@@ -63,6 +70,7 @@ func NewTaskService(db *gorm.DB, configs *OciConfigService, notifier *Notifier,
|
|||||||
settings: settings,
|
settings: settings,
|
||||||
cron: cron.New(),
|
cron: cron.New(),
|
||||||
entries: map[uint]cron.EntryID{},
|
entries: map[uint]cron.EntryID{},
|
||||||
|
running: map[uint]bool{},
|
||||||
snatchIPWait: defaultSnatchIPWait,
|
snatchIPWait: defaultSnatchIPWait,
|
||||||
snatchVars: map[uint]map[string]string{},
|
snatchVars: map[uint]map[string]string{},
|
||||||
}
|
}
|
||||||
@@ -86,10 +94,11 @@ func (s *TaskService) Start() error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop 停止调度并等待执行中的任务收尾,
|
// Stop 停止调度并等待执行中的任务收尾(含手动触发的后台执行),
|
||||||
// 保证任务产生的异步通知都已进入 Notifier 的等待队列。
|
// 保证任务产生的异步通知都已进入 Notifier 的等待队列。
|
||||||
func (s *TaskService) Stop() {
|
func (s *TaskService) Stop() {
|
||||||
<-s.cron.Stop().Done()
|
<-s.cron.Stop().Done()
|
||||||
|
s.runWG.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
// healthCheckPayload 是测活任务参数;ociConfigIds 为空表示全部配置。
|
// healthCheckPayload 是测活任务参数;ociConfigIds 为空表示全部配置。
|
||||||
@@ -356,7 +365,7 @@ func (s *TaskService) TaskLogs(ctx context.Context, id uint, limit int) ([]model
|
|||||||
return logs, nil
|
return logs, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// RunTaskNow 立即执行一次任务并返回本次日志。
|
// RunTaskNow 同步执行一次任务并返回本次日志(内部与测试使用;API 走 TriggerTask)。
|
||||||
func (s *TaskService) RunTaskNow(ctx context.Context, id uint) (*model.TaskLog, error) {
|
func (s *TaskService) RunTaskNow(ctx context.Context, id uint) (*model.TaskLog, error) {
|
||||||
if _, err := s.GetTask(ctx, id); err != nil {
|
if _, err := s.GetTask(ctx, id); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -364,6 +373,43 @@ func (s *TaskService) RunTaskNow(ctx context.Context, id uint) (*model.TaskLog,
|
|||||||
return s.execute(id), nil
|
return s.execute(id), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ErrTaskRunning 表示任务已有一次执行在途,拒绝重复触发。
|
||||||
|
var ErrTaskRunning = errors.New("任务正在执行中,请稍候")
|
||||||
|
|
||||||
|
// TriggerTask 异步触发一次任务执行并立即返回;执行结果照常落任务日志与通知,
|
||||||
|
// 由前端轮询呈现。同一任务在途时返回 ErrTaskRunning。
|
||||||
|
func (s *TaskService) TriggerTask(ctx context.Context, id uint) error {
|
||||||
|
if _, err := s.GetTask(ctx, id); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if !s.beginRun(id) {
|
||||||
|
return ErrTaskRunning
|
||||||
|
}
|
||||||
|
s.runWG.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer s.runWG.Done()
|
||||||
|
s.runHeld(id)
|
||||||
|
}()
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// beginRun 抢占任务的执行权;在途时返回 false。
|
||||||
|
func (s *TaskService) beginRun(id uint) bool {
|
||||||
|
s.runningMu.Lock()
|
||||||
|
defer s.runningMu.Unlock()
|
||||||
|
if s.running[id] {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
s.running[id] = true
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *TaskService) endRun(id uint) {
|
||||||
|
s.runningMu.Lock()
|
||||||
|
delete(s.running, id)
|
||||||
|
s.runningMu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
// schedule 把任务注册进 cron 调度。
|
// schedule 把任务注册进 cron 调度。
|
||||||
func (s *TaskService) schedule(task *model.Task) error {
|
func (s *TaskService) schedule(task *model.Task) error {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
@@ -412,9 +458,18 @@ func (s *TaskService) lockTenantCleanup() func() {
|
|||||||
return s.runMu.Unlock
|
return s.runMu.Unlock
|
||||||
}
|
}
|
||||||
|
|
||||||
// execute 执行一次任务:加载 → 分派 → 更新任务状态并写日志,
|
// execute 执行一次任务:同一任务已有执行在途时静默跳过(cron 重叠触发防抖)。
|
||||||
// 前后状态交给通知判定,只在状态变化时推送。
|
|
||||||
func (s *TaskService) execute(taskID uint) *model.TaskLog {
|
func (s *TaskService) execute(taskID uint) *model.TaskLog {
|
||||||
|
if !s.beginRun(taskID) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return s.runHeld(taskID)
|
||||||
|
}
|
||||||
|
|
||||||
|
// runHeld 在已持有执行权的前提下完成一次执行:加载 → 分派 → 更新任务状态并写日志,
|
||||||
|
// 前后状态交给通知判定,只在状态变化时推送。
|
||||||
|
func (s *TaskService) runHeld(taskID uint) *model.TaskLog {
|
||||||
|
defer s.endRun(taskID)
|
||||||
s.runMu.RLock()
|
s.runMu.RLock()
|
||||||
defer s.runMu.RUnlock()
|
defer s.runMu.RUnlock()
|
||||||
return s.executeLocked(taskID)
|
return s.executeLocked(taskID)
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package service
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -801,3 +802,36 @@ func TestNotifyEventsSnatchVarsMerged(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTriggerTaskAsyncAndDedup(t *testing.T) {
|
||||||
|
client := &blockingClient{fakeClient: &fakeClient{}, started: make(chan struct{}, 1), release: make(chan struct{})}
|
||||||
|
configs, tasks, db := newTenantDeleteEnv(t, client)
|
||||||
|
target, _ := seedDeleteTenants(t, db)
|
||||||
|
setTenantPrivateKey(t, configs, target.ID)
|
||||||
|
task := createHealthTask(t, tasks, target.ID)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
// 触发立即返回,执行在后台开始
|
||||||
|
if err := tasks.TriggerTask(ctx, task.ID); err != nil {
|
||||||
|
t.Fatalf("TriggerTask: %v", err)
|
||||||
|
}
|
||||||
|
<-client.started
|
||||||
|
// 在途重复触发被拒;cron 重叠执行静默跳过
|
||||||
|
if err := tasks.TriggerTask(ctx, task.ID); !errors.Is(err, ErrTaskRunning) {
|
||||||
|
t.Errorf("在途重复触发应返回 ErrTaskRunning: %v", err)
|
||||||
|
}
|
||||||
|
if entry := tasks.execute(task.ID); entry != nil {
|
||||||
|
t.Error("在途时 cron 重叠执行应静默跳过")
|
||||||
|
}
|
||||||
|
close(client.release)
|
||||||
|
tasks.Stop() // 等待后台执行收尾
|
||||||
|
assertCount(t, db, &model.TaskLog{}, "task_id = ?", []any{task.ID}, 1)
|
||||||
|
// 执行权已释放,可再次触发;不存在的任务报错
|
||||||
|
if !tasks.beginRun(task.ID) {
|
||||||
|
t.Error("执行结束后应可重新获得执行权")
|
||||||
|
}
|
||||||
|
tasks.endRun(task.ID)
|
||||||
|
if err := tasks.TriggerTask(ctx, 9999); err == nil {
|
||||||
|
t.Error("不存在的任务应报错")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"log"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
@@ -104,7 +105,8 @@ func planTenantTask(task model.Task, id uint) (tenantTaskAction, bool, error) {
|
|||||||
func planSnatchTask(task model.Task, id uint) (tenantTaskAction, bool, error) {
|
func planSnatchTask(task model.Task, id uint) (tenantTaskAction, bool, error) {
|
||||||
var payload snatchPayload
|
var payload snatchPayload
|
||||||
if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil {
|
if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil {
|
||||||
return tenantTaskAction{}, false, fmt.Errorf("parse task %d payload: %w", task.ID, err)
|
logSkippedTask(task, err)
|
||||||
|
return tenantTaskAction{}, false, nil
|
||||||
}
|
}
|
||||||
if payload.OciConfigID != id {
|
if payload.OciConfigID != id {
|
||||||
return tenantTaskAction{}, false, nil
|
return tenantTaskAction{}, false, nil
|
||||||
@@ -112,10 +114,17 @@ func planSnatchTask(task model.Task, id uint) (tenantTaskAction, bool, error) {
|
|||||||
return tenantTaskAction{task: task, deleteTask: true}, true, nil
|
return tenantTaskAction{task: task, deleteTask: true}, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// logSkippedTask 记录 payload 无法解析而被跳过的任务:坏数据无法归属租户,
|
||||||
|
// fail-closed 会永久阻断删除且只能手工修库;改为保留原任务(不删不改)并放行删除。
|
||||||
|
func logSkippedTask(task model.Task, err error) {
|
||||||
|
log.Printf("tenant delete: 任务 %d(%s)payload 无法解析,跳过处理: %v", task.ID, task.Type, err)
|
||||||
|
}
|
||||||
|
|
||||||
func planMultiTenantTask(task model.Task, id uint) (tenantTaskAction, bool, error) {
|
func planMultiTenantTask(task model.Task, id uint) (tenantTaskAction, bool, error) {
|
||||||
payload, err := decodeConfigIDs(task)
|
payload, err := decodeConfigIDs(task)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return tenantTaskAction{}, false, err
|
logSkippedTask(task, err)
|
||||||
|
return tenantTaskAction{}, false, nil
|
||||||
}
|
}
|
||||||
if len(payload.OciConfigIDs) == 0 {
|
if len(payload.OciConfigIDs) == 0 {
|
||||||
return tenantTaskAction{task: task, payload: task.Payload}, true, nil
|
return tenantTaskAction{task: task, payload: task.Payload}, true, nil
|
||||||
@@ -197,11 +206,11 @@ func resetTenantTask(tx *gorm.DB, action tenantTaskAction) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func deleteTenantEvents(tx *gorm.DB, id uint, result *tenantDeleteResult) error {
|
func deleteTenantEvents(tx *gorm.DB, id uint, result *tenantDeleteResult) error {
|
||||||
ruleIDs, eventIDs, affectedRules, err := loadTenantEventRefs(tx, id)
|
ruleIDs, affectedRules, err := loadTenantEventRefs(tx, id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := deleteAlertHits(tx, ruleIDs, eventIDs); err != nil {
|
if err := deleteAlertHits(tx, id); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
result.alertRuleIDs = mergeIDs(ruleIDs, affectedRules)
|
result.alertRuleIDs = mergeIDs(ruleIDs, affectedRules)
|
||||||
@@ -214,20 +223,19 @@ func deleteTenantEvents(tx *gorm.DB, id uint, result *tenantDeleteResult) error
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func loadTenantEventRefs(tx *gorm.DB, id uint) ([]uint, []uint, []uint, error) {
|
func loadTenantEventRefs(tx *gorm.DB, id uint) ([]uint, []uint, error) {
|
||||||
ruleIDs, err := lockedTenantRuleIDs(tx, id)
|
ruleIDs, err := lockedTenantRuleIDs(tx, id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, nil, fmt.Errorf("load tenant alert rules: %w", err)
|
return nil, nil, fmt.Errorf("load tenant alert rules: %w", err)
|
||||||
}
|
}
|
||||||
eventIDs, err := lockedTenantEventIDs(tx, id)
|
if err := lockTenantEventRows(tx, id); err != nil {
|
||||||
|
return nil, nil, fmt.Errorf("load tenant log events: %w", err)
|
||||||
|
}
|
||||||
|
affected, err := alertHitRuleIDs(tx, id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, nil, fmt.Errorf("load tenant log events: %w", err)
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
affected, err := alertHitRuleIDs(tx, eventIDs)
|
return ruleIDs, affected, nil
|
||||||
if err != nil {
|
|
||||||
return nil, nil, nil, err
|
|
||||||
}
|
|
||||||
return ruleIDs, eventIDs, affected, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func lockedTenantRuleIDs(tx *gorm.DB, id uint) ([]uint, error) {
|
func lockedTenantRuleIDs(tx *gorm.DB, id uint) ([]uint, error) {
|
||||||
@@ -241,23 +249,24 @@ func lockedTenantRuleIDs(tx *gorm.DB, id uint) ([]uint, error) {
|
|||||||
return ids, err
|
return ids, err
|
||||||
}
|
}
|
||||||
|
|
||||||
func lockedTenantEventIDs(tx *gorm.DB, id uint) ([]uint, error) {
|
// lockTenantEventRows 对租户全部日志事件行加 FOR UPDATE 锁(SQLite 忽略,
|
||||||
|
// MySQL/PG 阻塞并发命中写入);事件可达数万,ID 不再回传拼接 SQL,
|
||||||
|
// 后续删除与关联查询一律用子查询,避免绑定变量上限。
|
||||||
|
func lockTenantEventRows(tx *gorm.DB, id uint) error {
|
||||||
var rows []model.LogEvent
|
var rows []model.LogEvent
|
||||||
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").
|
return tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").
|
||||||
Where("oci_config_id = ?", id).Order("id").Find(&rows).Error
|
Where("oci_config_id = ?", id).Order("id").Find(&rows).Error
|
||||||
ids := make([]uint, 0, len(rows))
|
|
||||||
for _, row := range rows {
|
|
||||||
ids = append(ids, row.ID)
|
|
||||||
}
|
|
||||||
return ids, err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func alertHitRuleIDs(tx *gorm.DB, eventIDs []uint) ([]uint, error) {
|
// tenantEventIDs 构造「本租户日志事件 ID」子查询,供 IN (?) 内联。
|
||||||
if len(eventIDs) == 0 {
|
func tenantEventIDs(tx *gorm.DB, id uint) *gorm.DB {
|
||||||
return nil, nil
|
return tx.Model(&model.LogEvent{}).Select("id").Where("oci_config_id = ?", id)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// alertHitRuleIDs 找出命中引用了本租户事件的规则 ID(去重,可能含他租户/全局规则)。
|
||||||
|
func alertHitRuleIDs(tx *gorm.DB, id uint) ([]uint, error) {
|
||||||
ids := make([]uint, 0)
|
ids := make([]uint, 0)
|
||||||
err := tx.Model(&model.AlertRuleHit{}).Where("log_event_id IN ?", eventIDs).
|
err := tx.Model(&model.AlertRuleHit{}).Where("log_event_id IN (?)", tenantEventIDs(tx, id)).
|
||||||
Distinct().Pluck("rule_id", &ids).Error
|
Distinct().Pluck("rule_id", &ids).Error
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("load affected alert rules: %w", err)
|
return nil, fmt.Errorf("load affected alert rules: %w", err)
|
||||||
@@ -280,19 +289,13 @@ func mergeIDs(groups ...[]uint) []uint {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
func deleteAlertHits(tx *gorm.DB, ruleIDs, eventIDs []uint) error {
|
// deleteAlertHits 删除本租户相关的全部命中:规则属于本租户,或命中引用了
|
||||||
query := tx.Model(&model.AlertRuleHit{})
|
// 本租户的日志事件(他租户/全局规则命中本租户事件的行一并清)。
|
||||||
switch {
|
func deleteAlertHits(tx *gorm.DB, id uint) error {
|
||||||
case len(ruleIDs) > 0 && len(eventIDs) > 0:
|
rules := tx.Model(&model.AlertRule{}).Select("id").Where("oci_config_id = ?", id)
|
||||||
query = query.Where("rule_id IN ? OR log_event_id IN ?", ruleIDs, eventIDs)
|
err := tx.Where("rule_id IN (?) OR log_event_id IN (?)", rules, tenantEventIDs(tx, id)).
|
||||||
case len(ruleIDs) > 0:
|
Delete(&model.AlertRuleHit{}).Error
|
||||||
query = query.Where("rule_id IN ?", ruleIDs)
|
if err != nil {
|
||||||
case len(eventIDs) > 0:
|
|
||||||
query = query.Where("log_event_id IN ?", eventIDs)
|
|
||||||
default:
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if err := query.Delete(&model.AlertRuleHit{}).Error; err != nil {
|
|
||||||
return fmt.Errorf("delete tenant alert hits: %w", err)
|
return fmt.Errorf("delete tenant alert hits: %w", err)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -108,7 +108,10 @@ func tenantTaskCases() []tenantTaskCase {
|
|||||||
{name: "测活多租户", task: taskOf(model.TaskTypeHealthCheck, `{"ociConfigIds":[1,2]}`), wantOK: true, wantPayload: `{"ociConfigIds":[2]}`},
|
{name: "测活多租户", task: taskOf(model.TaskTypeHealthCheck, `{"ociConfigIds":[1,2]}`), wantOK: true, wantPayload: `{"ociConfigIds":[2]}`},
|
||||||
{name: "成本去重命中", task: taskOf(model.TaskTypeCost, `{"ociConfigIds":[1,1,2]}`), wantOK: true, wantPayload: `{"ociConfigIds":[2]}`},
|
{name: "成本去重命中", task: taskOf(model.TaskTypeCost, `{"ociConfigIds":[1,1,2]}`), wantOK: true, wantPayload: `{"ociConfigIds":[2]}`},
|
||||||
{name: "成本未命中", task: taskOf(model.TaskTypeCost, `{"ociConfigIds":[2]}`)},
|
{name: "成本未命中", task: taskOf(model.TaskTypeCost, `{"ociConfigIds":[2]}`)},
|
||||||
{name: "非法 JSON", task: taskOf(model.TaskTypeCost, `{`), wantErr: true},
|
// 坏 payload 记警告跳过(保留原任务),不再 fail-closed 阻断整个租户删除
|
||||||
|
{name: "非法 JSON 跳过", task: taskOf(model.TaskTypeCost, `{`)},
|
||||||
|
{name: "抢机坏 payload 跳过", task: taskOf(model.TaskTypeSnatch, `not-json`)},
|
||||||
|
{name: "抢机空 payload 跳过", task: taskOf(model.TaskTypeSnatch, ``)},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -607,3 +610,67 @@ func assertCount(t *testing.T, db *gorm.DB, value any, query string, args []any,
|
|||||||
t.Errorf("count %T = %d, want %d", value, got, want)
|
t.Errorf("count %T = %d, want %d", value, got, want)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDeleteTenantSkipsCorruptTaskPayload(t *testing.T) {
|
||||||
|
configs, _, db := newTenantDeleteEnv(t, &fakeClient{})
|
||||||
|
target, _ := seedDeleteTenants(t, db)
|
||||||
|
corrupt := model.Task{Name: "corrupt-snatch", Type: model.TaskTypeSnatch, Payload: `{broken`}
|
||||||
|
mustCreate(t, db, &corrupt)
|
||||||
|
mustCreate(t, db, &model.TaskLog{TaskID: corrupt.ID, Message: "keep"})
|
||||||
|
|
||||||
|
if err := configs.Delete(context.Background(), target.ID); err != nil {
|
||||||
|
t.Fatalf("坏 payload 不应阻断租户删除: %v", err)
|
||||||
|
}
|
||||||
|
var kept model.Task
|
||||||
|
if err := db.First(&kept, corrupt.ID).Error; err != nil || kept.Payload != `{broken` {
|
||||||
|
t.Errorf("坏任务应原样保留: %+v, %v", kept, err)
|
||||||
|
}
|
||||||
|
assertCount(t, db, &model.TaskLog{}, "task_id = ?", []any{corrupt.ID}, 1)
|
||||||
|
assertCount(t, db, &model.OciConfig{}, "id = ?", []any{target.ID}, 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeleteTenantEventsAffectedRules(t *testing.T) {
|
||||||
|
_, _, db := newTenantDeleteEnv(t, &fakeClient{})
|
||||||
|
target, other := seedDeleteTenants(t, db)
|
||||||
|
seedTenantEvents(t, db, target.ID, other.ID)
|
||||||
|
|
||||||
|
// 规则 ID 须在删除前取:target-rule 会随租户一并删除
|
||||||
|
var targetRule, globalRule model.AlertRule
|
||||||
|
db.Where("name = ?", "target-rule").First(&targetRule)
|
||||||
|
db.Where("name = ?", "global-rule").First(&globalRule)
|
||||||
|
|
||||||
|
result := &tenantDeleteResult{}
|
||||||
|
err := db.Transaction(func(tx *gorm.DB) error {
|
||||||
|
return deleteTenantEvents(tx, target.ID, result)
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("deleteTenantEvents: %v", err)
|
||||||
|
}
|
||||||
|
// target-rule 属本租户,global-rule 的命中引用了本租户事件:都应回收冷却
|
||||||
|
got := map[uint]bool{}
|
||||||
|
for _, id := range result.alertRuleIDs {
|
||||||
|
got[id] = true
|
||||||
|
}
|
||||||
|
if len(got) != 2 || !got[targetRule.ID] || !got[globalRule.ID] {
|
||||||
|
t.Errorf("alertRuleIDs = %v, want 含 target-rule(%d) 与 global-rule(%d)", result.alertRuleIDs, targetRule.ID, globalRule.ID)
|
||||||
|
}
|
||||||
|
// 仅保留 global-rule × other-event 一条命中
|
||||||
|
assertCount(t, db, &model.AlertRuleHit{}, "", nil, 1)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeleteTenantManyEventsNoVarLimit(t *testing.T) {
|
||||||
|
// 回归:事件数超 SQLite 绑定变量上限(32766)时删除仍成功(旧实现 IN 展开必失败)
|
||||||
|
configs, _, db := newTenantDeleteEnv(t, &fakeClient{})
|
||||||
|
target, _ := seedDeleteTenants(t, db)
|
||||||
|
events := make([]model.LogEvent, 0, 33000)
|
||||||
|
for i := 0; i < 33000; i++ {
|
||||||
|
events = append(events, model.LogEvent{OciConfigID: target.ID, MessageID: fmt.Sprintf("bulk-%d", i)})
|
||||||
|
}
|
||||||
|
if err := db.CreateInBatches(&events, 500).Error; err != nil {
|
||||||
|
t.Fatalf("seed events: %v", err)
|
||||||
|
}
|
||||||
|
if err := configs.Delete(context.Background(), target.ID); err != nil {
|
||||||
|
t.Fatalf("数万事件时删除不应受绑定变量上限影响: %v", err)
|
||||||
|
}
|
||||||
|
assertCount(t, db, &model.LogEvent{}, "oci_config_id = ?", []any{target.ID}, 0)
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user