diff --git a/pkg/handler/handler.go b/pkg/handler/handler.go index afdc12f..a8fc13d 100644 --- a/pkg/handler/handler.go +++ b/pkg/handler/handler.go @@ -262,6 +262,49 @@ func (h *Handler) StopInferenceTask(c *gin.Context) { c.JSON(http.StatusOK, response) } +// ReSubmitTask 处理重启推理任务请求 +// @Summary 重启推理任务 +// @Description 将入参原样转发给第三方 /jsm/v2/jobs/submit,返回新的 jobSetID 与入参中的 localJobID +// @Tags 任务控制 +// @Accept json +// @Produce json +// @Param request body models.SubtaskRequest true "重启任务参数(userID、jobSetInfo.jobs)" +// @Success 200 {object} models.ReSubmitTaskResponse +// @Failure 400 {object} Response +// @Failure 500 {object} Response +// @Router /api/v1/reSubmitTask [post] +func (h *Handler) ReSubmitTask(c *gin.Context) { + var request models.SubtaskRequest + if err := c.ShouldBindJSON(&request); err != nil { + c.JSON(http.StatusOK, BadRequestResponse("请求参数错误: "+err.Error())) + return + } + fmt.Printf("[ReSubmitTask] 请求参数: %+v\n", request) + + if len(request.JobSetInfo.Jobs) == 0 { + c.JSON(http.StatusOK, BadRequestResponse("jobSetInfo.jobs 不能为空")) + return + } + + // 获取认证信息 + authData, err := h.authService.GetToken(c.Request.Context()) + if err != nil { + c.JSON(http.StatusOK, InternalServerErrorResponse("获取认证信息失败: "+err.Error())) + return + } + + response, err := h.inferenceService.ReSubmitTask(c.Request.Context(), authData, &request) + if err != nil { + errResponse := InternalServerErrorResponse("重启任务失败: " + err.Error()) + fmt.Printf("[ReSubmitTask] 返回错误: %+v\n", errResponse) + c.JSON(http.StatusOK, errResponse) + return + } + + fmt.Printf("[ReSubmitTask] 返回结果: %+v\n", response) + c.JSON(http.StatusOK, response) +} + // SubmitInferenceTask 处理推理任务提交请求 // @Summary 提交推理任务(同步) // @Description 同步提交推理任务,包含代码、模型、增量模型的准备步骤,等待完成后返回结果 @@ -378,7 +421,7 @@ func (h *Handler) SubmitInferenceTaskAsync(c *gin.Context) { } response := models.InferenceAsyncSubmitResponse{ - Code: 200, + Code: http.StatusOK, Msg: "", Data: models.InferenceAsyncSubmitRespData{TaskID: taskID}, } diff --git a/pkg/handler/response.go b/pkg/handler/response.go index 730604c..8c9d6c8 100644 --- a/pkg/handler/response.go +++ b/pkg/handler/response.go @@ -1,5 +1,7 @@ package handler +import "net/http" + // Response 统一的API响应格式 type Response struct { Code int `json:"code"` // 状态码 @@ -10,7 +12,7 @@ type Response struct { // SuccessResponse 成功响应 func SuccessResponse(data interface{}) Response { return Response{ - Code: 200, + Code: http.StatusOK, Message: "success", Data: data, } diff --git a/pkg/models/task.go b/pkg/models/task.go index 6802504..98e79bb 100644 --- a/pkg/models/task.go +++ b/pkg/models/task.go @@ -322,6 +322,19 @@ type StopInferenceTaskResponse struct { Message string `json:"message"` // 消息 } +// ReSubmitTaskResponse 重启推理任务响应 +type ReSubmitTaskResponse struct { + Code int `json:"code"` // 状态码 + Message string `json:"message"` // 消息 + Data ReSubmitTaskData `json:"data"` // 响应数据 +} + +// ReSubmitTaskData 重启推理任务响应数据 +type ReSubmitTaskData struct { + JobSetID string `json:"jobSetID"` // 任务集ID(来自第三方接口返回) + LocalJobID string `json:"localJobID"` // 本地任务ID(来自入参 jobs) +} + // StopInferenceSubmitRequest 停止推理任务提交给第三方的请求 type StopInferenceSubmitRequest struct { UserID int `json:"userID"` // 用户ID diff --git a/pkg/router/router.go b/pkg/router/router.go index e0e2129..403e9bd 100644 --- a/pkg/router/router.go +++ b/pkg/router/router.go @@ -126,6 +126,7 @@ func SetupRouter() *gin.Engine { apiV1.POST("/submitTask", h.SubmitInferenceTask) apiV1.POST("/getTaskStatus", h.QueryStatus) apiV1.POST("/stopInferenceTask", h.StopInferenceTask) + apiV1.POST("/reSubmitTask", h.ReSubmitTask) // 异步推理任务提交/查询 apiV1.POST("/submitInferenceTaskAsync", h.SubmitInferenceTaskAsync) apiV1.POST("/getInferenceTaskAsync", h.GetInferenceTaskAsync) diff --git a/pkg/service/inference/inference_service.go b/pkg/service/inference/inference_service.go index 380358e..67761d1 100644 --- a/pkg/service/inference/inference_service.go +++ b/pkg/service/inference/inference_service.go @@ -5,6 +5,7 @@ import ( "database/sql" "encoding/json" "fmt" + "net/http" "os" "path/filepath" "remote-task-excutor-cli/pkg/client" @@ -77,7 +78,7 @@ func (s *inferenceService) GetTaskStatus(ctx context.Context, authData *models.A // 超时了,返回"启动中"状态,因为对端接口还在准备中 fmt.Printf("查询任务状态超时,返回启动中状态: %v\n", err) return &models.InferenceTaskStatusResponse{ - Code: 200, + Code: http.StatusOK, Msg: "", Data: models.InferenceTaskStatusData{ Status: "Init", @@ -107,7 +108,7 @@ func (s *inferenceService) GetTaskStatus(ctx context.Context, authData *models.A // 构建返回响应 return &models.InferenceTaskStatusResponse{ - Code: 200, + Code: http.StatusOK, Msg: "", Data: models.InferenceTaskStatusData{ Status: detailResp.Data.Instance.Status, @@ -169,11 +170,46 @@ func (s *inferenceService) StopTask(ctx context.Context, authData *models.AuthDa // 返回成功响应 return &models.StopInferenceTaskResponse{ - Code: 200, + Code: http.StatusOK, Message: "", }, nil } +// ReSubmitTask 重启推理任务:将入参原样转发给第三方 /jsm/v2/jobs/submit,返回 jobSetID 与入参中的 localJobID +func (s *inferenceService) ReSubmitTask(ctx context.Context, authData *models.AuthData, request *models.SubtaskRequest) (*models.ReSubmitTaskResponse, error) { + if request == nil || len(request.JobSetInfo.Jobs) == 0 { + return nil, fmt.Errorf("jobSetInfo.jobs 不能为空") + } + + s.httpClient.SetHeader("Authorization", "Bearer "+authData.Token) + + resp, err := s.httpClient.PostJSON("/jsm/v2/jobs/submit", request) + if err != nil { + return nil, fmt.Errorf("调用重启任务接口失败: %w", err) + } + + var submitResp models.SubmitTaskResponse + if err := json.Unmarshal(resp, &submitResp); err != nil { + return nil, fmt.Errorf("解析重启任务响应失败: %w", err) + } + + if submitResp.Code != models.ResponseOK { + return nil, fmt.Errorf("重启任务失败: %s - %s", submitResp.Code, submitResp.Message) + } + + localJobID := request.JobSetInfo.Jobs[0].LocalJobID + fmt.Printf("重启推理任务成功,JobSetID: %s, LocalJobID: %s\n", submitResp.Data.JobSetID, localJobID) + + return &models.ReSubmitTaskResponse{ + Code: http.StatusOK, + Message: "", + Data: models.ReSubmitTaskData{ + JobSetID: submitResp.Data.JobSetID, + LocalJobID: localJobID, + }, + }, nil +} + func (s *inferenceService) getLocalPath(path string) string { if s.apiCfg == nil { // 如果配置未设置,使用默认值 @@ -257,7 +293,7 @@ WHERE task_id = ? } return &models.InferenceAsyncTaskStatusResponse{ - Code: 200, + Code: http.StatusOK, Msg: "", Data: models.InferenceAsyncTaskStatusData{ TaskID: taskID, @@ -437,7 +473,7 @@ func (s *inferenceService) SubmitInferenceTask(ctx context.Context, authData *mo } return &models.InferenceSubmitTaskResponse{ - Code: 200, + Code: http.StatusOK, Msg: "", Data: taskData, }, nil diff --git a/pkg/service/interface.go b/pkg/service/interface.go index f4c1c81..b9b12f5 100644 --- a/pkg/service/interface.go +++ b/pkg/service/interface.go @@ -37,6 +37,7 @@ type InferenceService interface { GetInferenceTaskAsync(ctx context.Context, taskID string) (*models.InferenceAsyncTaskStatusResponse, error) GetTaskStatus(ctx context.Context, authData *models.AuthData, localJobID, jobSetID string) (*models.InferenceTaskStatusResponse, error) StopTask(ctx context.Context, authData *models.AuthData, localJobID, jobSetID string) (*models.StopInferenceTaskResponse, error) + ReSubmitTask(ctx context.Context, authData *models.AuthData, request *models.SubtaskRequest) (*models.ReSubmitTaskResponse, error) } type LogService interface {