diff --git a/config.yml b/config.yml index e7d2193..ef1c857 100644 --- a/config.yml +++ b/config.yml @@ -39,8 +39,8 @@ database: dryRun: false charset: "utf8" timezone: "Asia/Shanghai" - maxIdle: 5 - maxOpen: 20 + maxIdle: 15 + maxOpen: 60 maxLifetime: "30s" maxIdleConnTime: "30s" createdAt: "created_at" diff --git a/service/gateway/gateway_http_service.go b/service/gateway/gateway_http_service.go index efc3c35..3782545 100644 --- a/service/gateway/gateway_http_service.go +++ b/service/gateway/gateway_http_service.go @@ -7,7 +7,6 @@ import ( "errors" "fmt" "mime/multipart" - "model-gateway/common/util" "model-gateway/model/entity" "time" @@ -43,16 +42,25 @@ func UploadByTask(ctx context.Context, data []byte, fileExt string) (oss *Upload if err != nil { return nil, err } - if _, err := part.Write(data); err != nil { + if _, err = part.Write(data); err != nil { return nil, err } - contentType := writer.FormDataContentType() + //contentType := writer.FormDataContentType() if err = writer.Close(); err != nil { return nil, err } - headers := util.ForwardHeaders(ctx) - headers["Content-Type"] = contentType + //headers := util.ForwardHeaders(ctx) + //headers["Content-Type"] = contentType + + headers := make(map[string]string) + headers["Content-Type"] = writer.FormDataContentType() + if r := g.RequestFromCtx(ctx); r != nil { + if auth := r.Header.Get("Authorization"); auth != "" { + headers["Authorization"] = auth + } + } + fullURL := "oss/file/uploadFile" g.Log().Infof(ctx, "[OSS] upload start url=%s filename=%s size=%d", fullURL, filename, len(data)) @@ -78,15 +86,25 @@ type CallbackPayload struct { // TriggerCallback 任务的回调 func TriggerCallback(ctx context.Context, t *entity.ModelGatewayTask) { - headers := util.ForwardHeaders(ctx) + //headers := util.ForwardHeaders(ctx) + headers := make(map[string]string) + if r := g.RequestFromCtx(ctx); r != nil { + for k, v := range r.Request.Header { + if len(v) > 0 { + headers[k] = v[0] + } + } + } var resp struct{} payload := CallbackPayload{ TaskId: t.TaskID, State: t.State, - OssFile: t.ResultFile.OssFile, - FileType: t.ResultFile.FileType, ErrorMsg: t.ErrorMsg, } + if !g.IsEmpty(t.ResultFile) { + payload.OssFile = t.ResultFile.OssFile + payload.FileType = t.ResultFile.FileType + } jsonData, err := json.Marshal(payload) if err != nil { g.Log().Warningf(ctx, "[回调] JSON序列化失败 taskId=%s 错误=%v", t.TaskID, err) @@ -112,7 +130,15 @@ type PromptsCallbackPayload struct { // TriggerPromptsCallback 任务成功后的提示词回调 func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask, epicycleId int64) { callbackURL := "prompts-core/session/callback" - headers := util.ForwardHeaders(ctx) + //headers := util.ForwardHeaders(ctx) + headers := make(map[string]string) + if r := g.RequestFromCtx(ctx); r != nil { + for k, v := range r.Request.Header { + if len(v) > 0 { + headers[k] = v[0] + } + } + } var resp struct{} payload := PromptsCallbackPayload{ EpicycleId: epicycleId, @@ -136,7 +162,15 @@ func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask, epi // IsSuperAdmin 调用admin-go服务检查是否是超级管理员 func IsSuperAdmin(ctx context.Context) (res bool, err error) { - headers := util.ForwardHeaders(ctx) + //headers := util.ForwardHeaders(ctx) + headers := make(map[string]string) + if r := g.RequestFromCtx(ctx); r != nil { + for k, v := range r.Request.Header { + if len(v) > 0 { + headers[k] = v[0] + } + } + } var r = make(map[string]bool) if err = commonHttp.Get(ctx, "admin-go/api/v1/system/user/checkIsSuperAdmin", headers, &r); err != nil { return false, err diff --git a/service/task/task_service.go b/service/task/task_service.go index 6e5f710..b1d39f5 100644 --- a/service/task/task_service.go +++ b/service/task/task_service.go @@ -65,20 +65,20 @@ func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res * Body: req.RequestPayload, Headers: util.ParseHeadMsgHeaders(model.HeadMsg), } - id, err := dao.ModelGatewayTask.Insert(ctx, &entity.ModelGatewayTask{ - ModelName: req.ModelName, - TaskID: taskID, - State: public.TaskStatusPending, - BizName: req.BizName, - CallbackURL: req.CallbackUrl, - RequestPayload: &requestPayload, - EpicycleId: req.EpicycleId, - }) + task := new(entity.ModelGatewayTask) + task.ModelName = model.ModelName + task.TaskID = taskID + task.State = public.TaskStatusRunning + task.BizName = req.BizName + task.CallbackURL = req.CallbackUrl + task.RequestPayload = &requestPayload + task.EpicycleId = req.EpicycleId + id, err := dao.ModelGatewayTask.Insert(ctx, task) if err != nil { // 入库失败:回滚闸门占位 queue.ReleaseQueueSlot(ctx, req.ModelName, taskID) return nil, err } - + task.Id = id // 4) 写操作日志(不影响主流程,失败忽略) ip := "" ua := "" @@ -107,25 +107,25 @@ func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res * }, }) - // 5) 抢占任务:改为执行中 - rows, err := dao.ModelGatewayTask.Update(ctx, &entity.ModelGatewayTask{ - SQLBaseDO: beans.SQLBaseDO{Id: id}, - State: public.TaskStatusRunning, - }) - if err != nil { - return nil, err - } - if rows == 0 { - return nil, fmt.Errorf("任务不存在: id=%d", id) - } + //// 5) 抢占任务:改为执行中 + //rows, err := dao.ModelGatewayTask.Update(ctx, &entity.ModelGatewayTask{ + // SQLBaseDO: beans.SQLBaseDO{Id: id}, + // State: public.TaskStatusRunning, + //}) + //if err != nil { + // return nil, err + //} + //if rows == 0 { + // return nil, fmt.Errorf("任务不存在: id=%d", id) + //} // 6) 查询任务信息 - task, err := dao.ModelGatewayTask.Get(ctx, &entity.ModelGatewayTask{ - SQLBaseDO: beans.SQLBaseDO{Id: id}, - }) - if err != nil { - return nil, err - } + //task, err := dao.ModelGatewayTask.Get(ctx, &entity.ModelGatewayTask{ + // SQLBaseDO: beans.SQLBaseDO{Id: id}, + //}) + //if err != nil { + // return nil, err + //} // 7) 创建成功后立即异步尝试执行当前任务 go AsyncWorker.handleOne(util.AsyncCtx(ctx), task, model, req) diff --git a/service/task/worker.go b/service/task/worker.go index 2236c36..b74b91e 100644 --- a/service/task/worker.go +++ b/service/task/worker.go @@ -129,10 +129,14 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa if attempt > 0 { g.Log().Infof(ctx, "[执行任务][重试] OSS上传 第%d/%d次 taskId=%s", attempt, maxRetry, task.TaskID) } + startUpload := time.Now() oss, err = gateway.UploadByTask(ctx, gjson.New(result).MustToJson(), "json") if err == nil { break } + cost := time.Since(startUpload) + g.Log().Infof(ctx, "本次上传耗时:%s", cost) + g.Log().Errorf(ctx, "[执行任务][失败] OSS上传失败 taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, err) if attempt == maxRetry { task.State = public.TaskStatusFailed @@ -161,9 +165,9 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa } queue.ReleaseQueueSlot(ctx, task.ModelName, task.TaskID) - go gateway.TriggerCallback(context.WithoutCancel(ctx), task) + go gateway.TriggerCallback(util.AsyncCtx(ctx), task) if req.EpicycleId != 0 { - go gateway.TriggerPromptsCallback(context.WithoutCancel(ctx), task, req.EpicycleId) + go gateway.TriggerPromptsCallback(util.AsyncCtx(ctx), task, req.EpicycleId) } g.Log().Infof(ctx, "[执行任务][成功] taskId=%s duration=%ds fileType=%s", @@ -420,7 +424,7 @@ func injectErrorMessage(payload map[string]any, err error) map[string]any { // modelKey 用于覆盖/补充模型配置 head_msg(例如每次请求携带不同的 X-API-Key) func InvokeModel(ctx context.Context, model *entity.ModelGatewayModel, body map[string]any) ([]byte, error) { // 1) 记录模型调用次数 - _ = dao.ModelGatewayLogsStat.IncRequestCount(ctx, time.Now(), model.TenantId, model.Creator, model.ModelName) + //_ = dao.ModelGatewayLogsStat.IncRequestCount(ctx, time.Now(), model.TenantId, model.Creator, model.ModelName) // 2)请求参数映射:将标准 payload 按模型配置的 requestMapping 转为模型需要的格式 //—— 请求映射实际处理为提示词构建请求,因为有附加字段及其他字段的拼接。这里不方便做请求映射 @@ -447,13 +451,20 @@ func InvokeModel(ctx context.Context, model *entity.ModelGatewayModel, body map[ baseURL = baseURL + "?" + q.Encode() } } - req, err = http.NewRequestWithContext(ctx, http.MethodGet, baseURL, nil) + // 改用独立超时ctx,隔绝外层截止 + reqCtx, reqCancel := context.WithTimeout(context.Background(), timeout) + defer reqCancel() + req, err = http.NewRequestWithContext(reqCtx, http.MethodGet, baseURL, nil) + //req, err = http.NewRequestWithContext(ctx, http.MethodGet, baseURL, nil) default: bodyBytes, err := json.Marshal(body) if err != nil { return nil, err } - req, err = http.NewRequestWithContext(ctx, http.MethodPost, baseURL, bytes.NewReader(bodyBytes)) + reqCtx, reqCancel := context.WithTimeout(context.Background(), timeout) + defer reqCancel() + req, err = http.NewRequestWithContext(reqCtx, http.MethodPost, baseURL, bytes.NewReader(bodyBytes)) + //req, err = http.NewRequestWithContext(ctx, http.MethodPost, baseURL, bytes.NewReader(bodyBytes)) } // 5)注入请求头:先模型静态配置,再动态 modelKey(后者可覆盖前者) @@ -552,5 +563,5 @@ func (w *asyncWorker) failTask(ctx context.Context, t *entity.ModelGatewayTask, g.Log().Warningf(ctx, "[执行任务][更新数据库失败] taskId=%s err=%v", t.TaskID, err) } queue.ReleaseQueueSlot(ctx, t.ModelName, t.TaskID) - go gateway.TriggerCallback(context.WithoutCancel(ctx), t) + go gateway.TriggerCallback(util.AsyncCtx(ctx), t) }