Compare commits
12
Commits
a7c48dae84
...
5591b67f75
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5591b67f75 | ||
|
|
435ab285c1 | ||
|
|
8955e513aa | ||
|
|
f13377ca7d | ||
|
|
43329d2333 | ||
|
|
b90942de77 | ||
|
|
4fd331ebd8 | ||
|
|
d9ca9233b3 | ||
|
|
d79656c728 | ||
|
|
0a0e5fd7d4 | ||
|
|
22963d0e6f | ||
|
|
08ca4da691 |
@@ -3,7 +3,6 @@ package main
|
|||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
@@ -66,6 +65,8 @@ func jobSubmitViaAPI(t *testing.T, env *testenv.TestEnv, script string) int32 {
|
|||||||
return job.JobID
|
return job.JobID
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// [已弃用] 以下测试依赖 POST /api/v1/jobs/submit,该接口已被 POST /tasks 取代。
|
||||||
|
/*
|
||||||
// TestIntegration_Jobs_Submit verifies POST /api/v1/jobs/submit creates a new job.
|
// TestIntegration_Jobs_Submit verifies POST /api/v1/jobs/submit creates a new job.
|
||||||
func TestIntegration_Jobs_Submit(t *testing.T) {
|
func TestIntegration_Jobs_Submit(t *testing.T) {
|
||||||
env := testenv.NewTestEnv(t)
|
env := testenv.NewTestEnv(t)
|
||||||
@@ -220,3 +221,4 @@ func TestIntegration_Jobs_History(t *testing.T) {
|
|||||||
t.Fatalf("cancelled job %d not found in history", jobID)
|
t.Fatalf("cancelled job %d not found in history", jobID)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
*/
|
||||||
|
|||||||
@@ -53,7 +53,7 @@ func TestRouterRegistration(t *testing.T) {
|
|||||||
method string
|
method string
|
||||||
path string
|
path string
|
||||||
}{
|
}{
|
||||||
{"POST", "/api/v1/jobs/submit"},
|
// {"POST", "/api/v1/jobs/submit"}, // [已弃用] 已被 POST /tasks 取代
|
||||||
{"GET", "/api/v1/jobs"},
|
{"GET", "/api/v1/jobs"},
|
||||||
{"GET", "/api/v1/jobs/history"},
|
{"GET", "/api/v1/jobs/history"},
|
||||||
{"GET", "/api/v1/jobs/:id"},
|
{"GET", "/api/v1/jobs/:id"},
|
||||||
|
|||||||
+1
-120
@@ -42,80 +42,6 @@
|
|||||||
}
|
}
|
||||||
],
|
],
|
||||||
"paths": {
|
"paths": {
|
||||||
"/jobs/submit": {
|
|
||||||
"post": {
|
|
||||||
"tags": ["Jobs"],
|
|
||||||
"summary": "Submit a new job",
|
|
||||||
"description": "Submits a Slurm job with the specified script and optional parameters.",
|
|
||||||
"operationId": "submitJob",
|
|
||||||
"requestBody": {
|
|
||||||
"required": true,
|
|
||||||
"content": {
|
|
||||||
"application/json": {
|
|
||||||
"schema": {
|
|
||||||
"$ref": "#/components/schemas/SubmitJobRequest"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"responses": {
|
|
||||||
"201": {
|
|
||||||
"description": "Job submitted successfully",
|
|
||||||
"content": {
|
|
||||||
"application/json": {
|
|
||||||
"schema": {
|
|
||||||
"allOf": [
|
|
||||||
{ "$ref": "#/components/schemas/ApiResponseSuccess" },
|
|
||||||
{
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"data": {
|
|
||||||
"$ref": "#/components/schemas/JobResponse"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"400": {
|
|
||||||
"description": "Invalid request body or missing required fields",
|
|
||||||
"content": {
|
|
||||||
"application/json": {
|
|
||||||
"schema": {
|
|
||||||
"$ref": "#/components/schemas/ApiResponseError"
|
|
||||||
},
|
|
||||||
"examples": {
|
|
||||||
"invalid_body": {
|
|
||||||
"value": {
|
|
||||||
"success": false,
|
|
||||||
"error": "invalid request body"
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"missing_script": {
|
|
||||||
"value": {
|
|
||||||
"success": false,
|
|
||||||
"error": "script is required"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"502": {
|
|
||||||
"description": "Slurm backend error",
|
|
||||||
"content": {
|
|
||||||
"application/json": {
|
|
||||||
"schema": {
|
|
||||||
"$ref": "#/components/schemas/ApiResponseError"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"/jobs": {
|
"/jobs": {
|
||||||
"get": {
|
"get": {
|
||||||
"tags": ["Jobs"],
|
"tags": ["Jobs"],
|
||||||
@@ -2034,52 +1960,7 @@
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"SubmitJobRequest": {
|
|
||||||
"type": "object",
|
|
||||||
"required": ["script"],
|
|
||||||
"properties": {
|
|
||||||
"script": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Job script content"
|
|
||||||
},
|
|
||||||
"partition": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Target partition"
|
|
||||||
},
|
|
||||||
"qos": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Quality of Service"
|
|
||||||
},
|
|
||||||
"cpus": {
|
|
||||||
"type": "integer",
|
|
||||||
"format": "int32",
|
|
||||||
"description": "Number of CPUs required"
|
|
||||||
},
|
|
||||||
"memory": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Memory requirement (e.g. \"4G\")"
|
|
||||||
},
|
|
||||||
"time_limit": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Time limit (e.g. \"1:00:00\")"
|
|
||||||
},
|
|
||||||
"job_name": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Job name"
|
|
||||||
},
|
|
||||||
"environment": {
|
|
||||||
"type": "object",
|
|
||||||
"additionalProperties": {
|
|
||||||
"type": "string"
|
|
||||||
},
|
|
||||||
"description": "Environment variables"
|
|
||||||
},
|
|
||||||
"work_dir": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Job working directory"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"JobResponse": {
|
"JobResponse": {
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"properties": {
|
"properties": {
|
||||||
|
|||||||
@@ -8,12 +8,15 @@ import (
|
|||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TaskPollable defines the interface for refreshing stale task statuses.
|
// TaskPollable defines the interface for refreshing stale task statuses
|
||||||
|
// and recovering stuck tasks.
|
||||||
type TaskPollable interface {
|
type TaskPollable interface {
|
||||||
RefreshStaleTasks(ctx context.Context) error
|
RefreshStaleTasks(ctx context.Context) error
|
||||||
|
RecoverStuckTasks(ctx context.Context)
|
||||||
}
|
}
|
||||||
|
|
||||||
// TaskPoller periodically polls Slurm for task status updates via TaskPollable.
|
// TaskPoller periodically polls Slurm for task status updates and recovers
|
||||||
|
// stuck tasks via TaskPollable.
|
||||||
type TaskPoller struct {
|
type TaskPoller struct {
|
||||||
taskSvc TaskPollable
|
taskSvc TaskPollable
|
||||||
interval time.Duration
|
interval time.Duration
|
||||||
@@ -31,9 +34,11 @@ func NewTaskPoller(taskSvc TaskPollable, interval time.Duration, logger *zap.Log
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start launches the background goroutine that periodically refreshes stale tasks.
|
// Start launches background goroutines that periodically refresh stale tasks
|
||||||
|
// and recover stuck tasks.
|
||||||
func (p *TaskPoller) Start(ctx context.Context) {
|
func (p *TaskPoller) Start(ctx context.Context) {
|
||||||
ctx, p.cancel = context.WithCancel(ctx)
|
ctx, p.cancel = context.WithCancel(ctx)
|
||||||
|
|
||||||
p.wg.Add(1)
|
p.wg.Add(1)
|
||||||
go func() {
|
go func() {
|
||||||
defer p.wg.Done()
|
defer p.wg.Done()
|
||||||
@@ -50,9 +55,24 @@ func (p *TaskPoller) Start(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
p.wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer p.wg.Done()
|
||||||
|
ticker := time.NewTicker(5 * time.Minute)
|
||||||
|
defer ticker.Stop()
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
p.taskSvc.RecoverStuckTasks(ctx)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop cancels the background goroutine and waits for it to finish.
|
// Stop cancels the background goroutines and waits for them to finish.
|
||||||
func (p *TaskPoller) Stop() {
|
func (p *TaskPoller) Stop() {
|
||||||
if p.cancel != nil {
|
if p.cancel != nil {
|
||||||
p.cancel()
|
p.cancel()
|
||||||
|
|||||||
@@ -25,6 +25,8 @@ func (m *mockTaskPollable) RefreshStaleTasks(ctx context.Context) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *mockTaskPollable) RecoverStuckTasks(ctx context.Context) {}
|
||||||
|
|
||||||
func (m *mockTaskPollable) getCallCount() int {
|
func (m *mockTaskPollable) getCallCount() int {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
defer m.mu.Unlock()
|
defer m.mu.Unlock()
|
||||||
|
|||||||
+24
-22
@@ -22,29 +22,31 @@ func NewJobHandler(jobSvc *service.JobService, logger *zap.Logger) *JobHandler {
|
|||||||
return &JobHandler{jobSvc: jobSvc, logger: logger}
|
return &JobHandler{jobSvc: jobSvc, logger: logger}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// [已弃用] SubmitJob 已被 POST /tasks 取代。
|
||||||
|
// 保留方法体以防需要回滚。
|
||||||
// SubmitJob handles POST /api/v1/jobs/submit.
|
// SubmitJob handles POST /api/v1/jobs/submit.
|
||||||
func (h *JobHandler) SubmitJob(c *gin.Context) {
|
// func (h *JobHandler) SubmitJob(c *gin.Context) {
|
||||||
var req model.SubmitJobRequest
|
// var req model.SubmitJobRequest
|
||||||
if err := c.ShouldBindJSON(&req); err != nil {
|
// if err := c.ShouldBindJSON(&req); err != nil {
|
||||||
h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "invalid request body"))
|
// h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "invalid request body"))
|
||||||
server.BadRequest(c, "invalid request body")
|
// server.BadRequest(c, "invalid request body")
|
||||||
return
|
// return
|
||||||
}
|
// }
|
||||||
if req.Script == "" {
|
// if req.Script == "" {
|
||||||
h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "script is required"))
|
// h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "script is required"))
|
||||||
server.BadRequest(c, "script is required")
|
// server.BadRequest(c, "script is required")
|
||||||
return
|
// return
|
||||||
}
|
// }
|
||||||
|
//
|
||||||
resp, err := h.jobSvc.SubmitJob(c.Request.Context(), &req)
|
// resp, err := h.jobSvc.SubmitJob(c.Request.Context(), &req)
|
||||||
if err != nil {
|
// if err != nil {
|
||||||
h.logger.Error("handler error", zap.String("method", "SubmitJob"), zap.Int("status", http.StatusBadGateway), zap.Error(err))
|
// h.logger.Error("handler error", zap.String("method", "SubmitJob"), zap.Int("status", http.StatusBadGateway), zap.Error(err))
|
||||||
server.ErrorWithStatus(c, http.StatusBadGateway, "slurm error: "+err.Error())
|
// server.ErrorWithStatus(c, http.StatusBadGateway, "slurm error: "+err.Error())
|
||||||
return
|
// return
|
||||||
}
|
// }
|
||||||
|
//
|
||||||
server.Created(c, resp)
|
// server.Created(c, resp)
|
||||||
}
|
// }
|
||||||
|
|
||||||
// GetJobs handles GET /api/v1/jobs with pagination.
|
// GetJobs handles GET /api/v1/jobs with pagination.
|
||||||
func (h *JobHandler) GetJobs(c *gin.Context) {
|
func (h *JobHandler) GetJobs(c *gin.Context) {
|
||||||
|
|||||||
@@ -1,9 +1,7 @@
|
|||||||
package handler
|
package handler
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -23,7 +21,7 @@ func setupJobRouter(h *JobHandler) *gin.Engine {
|
|||||||
v1 := r.Group("/api/v1")
|
v1 := r.Group("/api/v1")
|
||||||
jobs := v1.Group("/jobs")
|
jobs := v1.Group("/jobs")
|
||||||
{
|
{
|
||||||
jobs.POST("/submit", h.SubmitJob)
|
// jobs.POST("/submit", h.SubmitJob) // [已弃用] 已被 POST /tasks 取代
|
||||||
jobs.GET("", h.GetJobs)
|
jobs.GET("", h.GetJobs)
|
||||||
jobs.GET("/history", h.GetJobHistory)
|
jobs.GET("/history", h.GetJobHistory)
|
||||||
jobs.GET("/:id", h.GetJob)
|
jobs.GET("/:id", h.GetJob)
|
||||||
@@ -61,6 +59,8 @@ func handlerLogs(logs *observer.ObservedLogs) []observer.LoggedEntry {
|
|||||||
return handler
|
return handler
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// [已弃用] SubmitJob 相关测试已被禁用,该接口已被 POST /tasks 取代。
|
||||||
|
/*
|
||||||
func TestSubmitJob_Success(t *testing.T) {
|
func TestSubmitJob_Success(t *testing.T) {
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
mux.HandleFunc("/slurm/v0.0.40/job/submit", func(w http.ResponseWriter, r *http.Request) {
|
mux.HandleFunc("/slurm/v0.0.40/job/submit", func(w http.ResponseWriter, r *http.Request) {
|
||||||
@@ -171,6 +171,9 @@ func TestSubmitJob_SlurmError(t *testing.T) {
|
|||||||
t.Fatalf("expected 502, got %d: %s", w.Code, w.Body.String())
|
t.Fatalf("expected 502, got %d: %s", w.Code, w.Body.String())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
*/
|
||||||
|
|
||||||
|
// --- Logging verification tests ---
|
||||||
|
|
||||||
func TestGetJobs_Success(t *testing.T) {
|
func TestGetJobs_Success(t *testing.T) {
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
@@ -462,6 +465,7 @@ func TestGetJobHistory_DefaultPagination(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
func TestSubmitJob_InvalidBody(t *testing.T) {
|
func TestSubmitJob_InvalidBody(t *testing.T) {
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
srv, handler := setupJobHandler(mux)
|
srv, handler := setupJobHandler(mux)
|
||||||
@@ -479,9 +483,11 @@ func TestSubmitJob_InvalidBody(t *testing.T) {
|
|||||||
t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String())
|
t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
*/
|
||||||
|
|
||||||
// --- Logging verification tests ---
|
// --- Logging verification tests ---
|
||||||
|
|
||||||
|
/*
|
||||||
func TestSubmitJob_InvalidBody_LogsWarn(t *testing.T) {
|
func TestSubmitJob_InvalidBody_LogsWarn(t *testing.T) {
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
srv, handler, logs := setupJobHandlerWithObserver(mux)
|
srv, handler, logs := setupJobHandlerWithObserver(mux)
|
||||||
@@ -614,6 +620,7 @@ func TestSubmitJob_Success_NoHandlerLogs(t *testing.T) {
|
|||||||
t.Errorf("expected no handler log entries on success, got %d", len(hLogs))
|
t.Errorf("expected no handler log entries on success, got %d", len(hLogs))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
*/
|
||||||
|
|
||||||
func TestGetJobs_Error_LogsError(t *testing.T) {
|
func TestGetJobs_Error_LogsError(t *testing.T) {
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ type Application struct {
|
|||||||
Category string `gorm:"size:255" json:"category,omitempty"` // 分类
|
Category string `gorm:"size:255" json:"category,omitempty"` // 分类
|
||||||
ScriptTemplate string `gorm:"type:text;not null" json:"script_template"` // 脚本模板
|
ScriptTemplate string `gorm:"type:text;not null" json:"script_template"` // 脚本模板
|
||||||
Parameters json.RawMessage `gorm:"type:json" json:"parameters,omitempty"` // 参数表单JSON
|
Parameters json.RawMessage `gorm:"type:json" json:"parameters,omitempty"` // 参数表单JSON
|
||||||
|
Environment json.RawMessage `gorm:"type:text" json:"environment,omitempty"` // 环境变量JSON
|
||||||
Scope string `gorm:"size:50;default:'system'" json:"scope,omitempty"` // 作用域(system/user)
|
Scope string `gorm:"size:50;default:'system'" json:"scope,omitempty"` // 作用域(system/user)
|
||||||
CreatedBy int64 `json:"created_by,omitempty"` // 创建者ID
|
CreatedBy int64 `json:"created_by,omitempty"` // 创建者ID
|
||||||
CreatedAt time.Time `json:"created_at"` // 创建时间
|
CreatedAt time.Time `json:"created_at"` // 创建时间
|
||||||
@@ -43,6 +44,7 @@ type ParameterSchema struct {
|
|||||||
Default string `json:"default,omitempty"` // 默认值
|
Default string `json:"default,omitempty"` // 默认值
|
||||||
Options []string `json:"options,omitempty"` // 枚举选项列表
|
Options []string `json:"options,omitempty"` // 枚举选项列表
|
||||||
Description string `json:"description,omitempty"` // 参数说明
|
Description string `json:"description,omitempty"` // 参数说明
|
||||||
|
SchedulingMap string `json:"scheduling_map,omitempty"` // maps to a scheduling param
|
||||||
}
|
}
|
||||||
|
|
||||||
// CreateApplicationRequest 是创建应用的 API 请求。
|
// CreateApplicationRequest 是创建应用的 API 请求。
|
||||||
@@ -53,6 +55,7 @@ type CreateApplicationRequest struct {
|
|||||||
Category string `json:"category,omitempty"` // 分类
|
Category string `json:"category,omitempty"` // 分类
|
||||||
ScriptTemplate string `json:"script_template" binding:"required"` // 脚本模板(必填)
|
ScriptTemplate string `json:"script_template" binding:"required"` // 脚本模板(必填)
|
||||||
Parameters json.RawMessage `json:"parameters,omitempty"` // 参数表单JSON
|
Parameters json.RawMessage `json:"parameters,omitempty"` // 参数表单JSON
|
||||||
|
Environment map[string]string `json:"environment,omitempty"` // 环境变量
|
||||||
Scope string `json:"scope,omitempty"` // 作用域
|
Scope string `json:"scope,omitempty"` // 作用域
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -64,6 +67,7 @@ type UpdateApplicationRequest struct {
|
|||||||
Category *string `json:"category,omitempty"` // 分类
|
Category *string `json:"category,omitempty"` // 分类
|
||||||
ScriptTemplate *string `json:"script_template,omitempty"` // 脚本模板
|
ScriptTemplate *string `json:"script_template,omitempty"` // 脚本模板
|
||||||
Parameters *json.RawMessage `json:"parameters,omitempty"` // 参数表单JSON
|
Parameters *json.RawMessage `json:"parameters,omitempty"` // 参数表单JSON
|
||||||
|
Environment *json.RawMessage `json:"environment,omitempty"` // 环境变量
|
||||||
Scope *string `json:"scope,omitempty"` // 作用域
|
Scope *string `json:"scope,omitempty"` // 作用域
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -57,15 +57,27 @@ type JobResponse struct {
|
|||||||
|
|
||||||
// Resources
|
// Resources
|
||||||
Cpus *int32 `json:"cpus,omitempty"` // 分配/请求的 CPU 核数
|
Cpus *int32 `json:"cpus,omitempty"` // 分配/请求的 CPU 核数
|
||||||
|
CpusPerTask *int32 `json:"cpus_per_task,omitempty"` // 每任务 CPU 核数
|
||||||
Tasks *int32 `json:"tasks,omitempty"` // 任务数
|
Tasks *int32 `json:"tasks,omitempty"` // 任务数
|
||||||
NodeCount *int32 `json:"node_count,omitempty"` // 节点数
|
NodeCount *int32 `json:"node_count,omitempty"` // 节点数
|
||||||
Nodes string `json:"nodes,omitempty"` // 分配的节点列表
|
Nodes string `json:"nodes,omitempty"` // 分配的节点列表
|
||||||
BatchHost string `json:"batch_host,omitempty"` // 批处理主节点
|
BatchHost string `json:"batch_host,omitempty"` // 批处理主节点
|
||||||
|
MemoryPerCpu *int64 `json:"memory_per_cpu,omitempty"` // 每 CPU 内存 (MB)
|
||||||
|
MemoryPerNode *int64 `json:"memory_per_node,omitempty"` // 每节点内存 (MB)
|
||||||
|
MemoryUsed *int64 `json:"memory_used,omitempty"` // 实际峰值内存消耗 (MB,仅历史作业)
|
||||||
|
|
||||||
|
// TRES (Trackable Resources)
|
||||||
|
TresReqStr string `json:"tres_req_str,omitempty"` // 请求的 TRES 字符串 (活跃作业)
|
||||||
|
TresAllocStr string `json:"tres_alloc_str,omitempty"` // 分配的 TRES 字符串 (活跃作业)
|
||||||
|
TresRequested string `json:"tres_requested,omitempty"` // 请求的 TRES (历史作业)
|
||||||
|
TresAllocated string `json:"tres_allocated,omitempty"` // 分配的 TRES (历史作业,最接近实际消耗)
|
||||||
|
GresDetail []string `json:"gres_detail,omitempty"` // GPU/GRES 详情
|
||||||
|
|
||||||
// Timing (Unix timestamp)
|
// Timing (Unix timestamp)
|
||||||
SubmitTime *int64 `json:"submit_time,omitempty"` // 提交时间
|
SubmitTime *int64 `json:"submit_time,omitempty"` // 提交时间
|
||||||
StartTime *int64 `json:"start_time,omitempty"` // 开始运行时间
|
StartTime *int64 `json:"start_time,omitempty"` // 开始运行时间
|
||||||
EndTime *int64 `json:"end_time,omitempty"` // 结束/预计结束时间
|
EndTime *int64 `json:"end_time,omitempty"` // 结束/预计结束时间
|
||||||
|
Elapsed *int32 `json:"elapsed,omitempty"` // 运行时长 (秒,仅历史作业)
|
||||||
|
|
||||||
// Result
|
// Result
|
||||||
ExitCode *int32 `json:"exit_code,omitempty"` // 退出码 (nil 表示未结束)
|
ExitCode *int32 `json:"exit_code,omitempty"` // 退出码 (nil 表示未结束)
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type JobHandler interface {
|
type JobHandler interface {
|
||||||
SubmitJob(c *gin.Context)
|
// SubmitJob(c *gin.Context) // [已弃用] 已被 POST /tasks 取代
|
||||||
GetJobs(c *gin.Context)
|
GetJobs(c *gin.Context)
|
||||||
GetJobHistory(c *gin.Context)
|
GetJobHistory(c *gin.Context)
|
||||||
GetJob(c *gin.Context)
|
GetJob(c *gin.Context)
|
||||||
@@ -73,7 +73,7 @@ func NewRouter(jobH JobHandler, clusterH ClusterHandler, appH ApplicationHandler
|
|||||||
v1 := r.Group("/api/v1")
|
v1 := r.Group("/api/v1")
|
||||||
|
|
||||||
jobs := v1.Group("/jobs")
|
jobs := v1.Group("/jobs")
|
||||||
jobs.POST("/submit", jobH.SubmitJob)
|
// jobs.POST("/submit", jobH.SubmitJob) // [已弃用] 已被 POST /tasks 取代
|
||||||
jobs.GET("", jobH.GetJobs)
|
jobs.GET("", jobH.GetJobs)
|
||||||
jobs.GET("/history", jobH.GetJobHistory)
|
jobs.GET("/history", jobH.GetJobHistory)
|
||||||
jobs.GET("/:id", jobH.GetJob)
|
jobs.GET("/:id", jobH.GetJob)
|
||||||
@@ -144,7 +144,7 @@ func NewTestRouter() *gin.Engine {
|
|||||||
|
|
||||||
func registerPlaceholderRoutes(v1 *gin.RouterGroup) {
|
func registerPlaceholderRoutes(v1 *gin.RouterGroup) {
|
||||||
jobs := v1.Group("/jobs")
|
jobs := v1.Group("/jobs")
|
||||||
jobs.POST("/submit", notImplemented)
|
// jobs.POST("/submit", notImplemented) // [已弃用] 已被 POST /tasks 取代
|
||||||
jobs.GET("", notImplemented)
|
jobs.GET("", notImplemented)
|
||||||
jobs.GET("/history", notImplemented)
|
jobs.GET("/history", notImplemented)
|
||||||
jobs.GET("/:id", notImplemented)
|
jobs.GET("/:id", notImplemented)
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ func TestAllRoutesRegistered(t *testing.T) {
|
|||||||
method string
|
method string
|
||||||
path string
|
path string
|
||||||
}{
|
}{
|
||||||
{"POST", "/api/v1/jobs/submit"},
|
// {"POST", "/api/v1/jobs/submit"}, // [已弃用] 已被 POST /tasks 取代
|
||||||
{"GET", "/api/v1/jobs"},
|
{"GET", "/api/v1/jobs"},
|
||||||
{"GET", "/api/v1/jobs/history"},
|
{"GET", "/api/v1/jobs/history"},
|
||||||
{"GET", "/api/v1/jobs/:id"},
|
{"GET", "/api/v1/jobs/:id"},
|
||||||
|
|||||||
@@ -42,6 +42,13 @@ func derefInt32ToStr(i *int32) string {
|
|||||||
return strconv.FormatInt(int64(*i), 10)
|
return strconv.FormatInt(int64(*i), 10)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func derefInt64ToStr(i *int64) string {
|
||||||
|
if i == nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return strconv.FormatInt(*i, 10)
|
||||||
|
}
|
||||||
|
|
||||||
func uint32NoValString(v *slurm.Uint32NoVal) string {
|
func uint32NoValString(v *slurm.Uint32NoVal) string {
|
||||||
if v == nil {
|
if v == nil {
|
||||||
return ""
|
return ""
|
||||||
|
|||||||
@@ -444,6 +444,26 @@ func TestClusterService_GetPartition_ErrorLogging(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDerefInt64ToStr(t *testing.T) {
|
||||||
|
t.Run("nil returns empty", func(t *testing.T) {
|
||||||
|
if got := derefInt64ToStr(nil); got != "" {
|
||||||
|
t.Errorf("derefInt64ToStr(nil) = %q, want empty", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
t.Run("non-nil returns string", func(t *testing.T) {
|
||||||
|
v := int64(4096)
|
||||||
|
if got := derefInt64ToStr(&v); got != "4096" {
|
||||||
|
t.Errorf("derefInt64ToStr(4096) = %q, want %q", got, "4096")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
t.Run("zero value", func(t *testing.T) {
|
||||||
|
v := int64(0)
|
||||||
|
if got := derefInt64ToStr(&v); got != "0" {
|
||||||
|
t.Errorf("derefInt64ToStr(0) = %q, want %q", got, "0")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
func TestClusterService_GetDiag_ErrorLogging(t *testing.T) {
|
func TestClusterService_GetDiag_ErrorLogging(t *testing.T) {
|
||||||
srv := errorServer()
|
srv := errorServer()
|
||||||
defer srv.Close()
|
defer srv.Close()
|
||||||
|
|||||||
@@ -37,7 +37,7 @@ func (s *JobService) SubmitJob(ctx context.Context, req *model.SubmitJobRequest)
|
|||||||
jobDesc.CurrentWorkingDirectory = &req.WorkDir
|
jobDesc.CurrentWorkingDirectory = &req.WorkDir
|
||||||
}
|
}
|
||||||
if req.CPUs > 0 {
|
if req.CPUs > 0 {
|
||||||
jobDesc.MinimumCpus = slurm.Ptr(req.CPUs)
|
jobDesc.CpusPerTask = slurm.Ptr(req.CPUs)
|
||||||
}
|
}
|
||||||
if req.TimeLimit != "" {
|
if req.TimeLimit != "" {
|
||||||
if mins, err := strconv.ParseInt(req.TimeLimit, 10, 64); err == nil {
|
if mins, err := strconv.ParseInt(req.TimeLimit, 10, 64); err == nil {
|
||||||
@@ -49,12 +49,15 @@ func (s *JobService) SubmitJob(ctx context.Context, req *model.SubmitJobRequest)
|
|||||||
"PATH=/usr/local/bin:/usr/bin:/bin",
|
"PATH=/usr/local/bin:/usr/bin:/bin",
|
||||||
"HOME=/root",
|
"HOME=/root",
|
||||||
}
|
}
|
||||||
|
for k, v := range req.Environment {
|
||||||
|
jobDesc.Environment = append(jobDesc.Environment, k+"="+v)
|
||||||
|
}
|
||||||
|
|
||||||
if req.MemoryPerNode != nil {
|
if req.MemoryPerNode != nil {
|
||||||
jobDesc.MemoryPerNode = &slurm.Uint64NoVal{Number: req.MemoryPerNode}
|
jobDesc.MemoryPerNode = &slurm.Uint64NoVal{Set: slurm.Ptr(true), Number: req.MemoryPerNode}
|
||||||
}
|
}
|
||||||
if req.MemoryPerCpu != nil {
|
if req.MemoryPerCpu != nil {
|
||||||
jobDesc.MemoryPerCpu = &slurm.Uint64NoVal{Number: req.MemoryPerCpu}
|
jobDesc.MemoryPerCpu = &slurm.Uint64NoVal{Set: slurm.Ptr(true), Number: req.MemoryPerCpu}
|
||||||
}
|
}
|
||||||
if req.Nodes != nil {
|
if req.Nodes != nil {
|
||||||
jobDesc.Nodes = req.Nodes
|
jobDesc.Nodes = req.Nodes
|
||||||
@@ -461,6 +464,88 @@ func mapUint32NoValToInt32(v *slurm.Uint32NoVal) *int32 {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func mapUint16NoValToInt32(v *slurm.Uint16NoVal) *int32 {
|
||||||
|
if v != nil && v.Number != nil {
|
||||||
|
n := int32(*v.Number)
|
||||||
|
return &n
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func mapUint64NoValToInt64(v *slurm.Uint64NoVal) *int64 {
|
||||||
|
if v != nil && v.Number != nil && *v.Number != 0 {
|
||||||
|
return v.Number
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// formatTresList serializes a TresList to Slurm-style TRES string.
|
||||||
|
// Format: "type=count,type:name=count" (e.g. "billing=4,cpu=4,mem=16384M").
|
||||||
|
func formatTresList(tl slurm.TresList) string {
|
||||||
|
if len(tl) == 0 {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
parts := make([]string, 0, len(tl))
|
||||||
|
for _, t := range tl {
|
||||||
|
if t.Type == nil || t.Count == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
key := *t.Type
|
||||||
|
if t.Name != nil && *t.Name != "" {
|
||||||
|
key += ":" + *t.Name
|
||||||
|
}
|
||||||
|
parts = append(parts, fmt.Sprintf("%s=%d", key, *t.Count))
|
||||||
|
}
|
||||||
|
return strings.Join(parts, ",")
|
||||||
|
}
|
||||||
|
|
||||||
|
// parseGresDetail parses a comma-separated GRES string into a slice.
|
||||||
|
func parseGresDetail(s string) []string {
|
||||||
|
if s == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
parts := strings.Split(s, ",")
|
||||||
|
result := make([]string, 0, len(parts))
|
||||||
|
for _, p := range parts {
|
||||||
|
p = strings.TrimSpace(p)
|
||||||
|
if p != "" {
|
||||||
|
result = append(result, p)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(result) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
// extractMemoryFromSteps extracts peak memory usage (in MB) from a SlurmDBD
|
||||||
|
// job's step TRES data. Despite the confusing naming, Slurm stores actual
|
||||||
|
// resource consumption (CPU time, memory RSS) in Tres.Requested.Max, not in
|
||||||
|
// Tres.Consumed (which holds I/O output data like disk/network).
|
||||||
|
// It scans ALL steps and returns the maximum mem value found.
|
||||||
|
// Returns nil if no memory data is available.
|
||||||
|
func extractMemoryFromSteps(steps slurm.StepList) *int64 {
|
||||||
|
var peakMB int64
|
||||||
|
for _, step := range steps {
|
||||||
|
if step.Tres == nil || step.Tres.Requested == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for _, t := range step.Tres.Requested.Max {
|
||||||
|
if t.Type != nil && *t.Type == "mem" && t.Count != nil {
|
||||||
|
// Slurm reports memory in bytes, convert to MB
|
||||||
|
mb := *t.Count / (1024 * 1024)
|
||||||
|
if mb > peakMB {
|
||||||
|
peakMB = mb
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if peakMB > 0 {
|
||||||
|
return &peakMB
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// mapJobInfo maps SDK JobInfo to API JobResponse.
|
// mapJobInfo maps SDK JobInfo to API JobResponse.
|
||||||
func mapJobInfo(ji *slurm.JobInfo) model.JobResponse {
|
func mapJobInfo(ji *slurm.JobInfo) model.JobResponse {
|
||||||
resp := model.JobResponse{}
|
resp := model.JobResponse{}
|
||||||
@@ -485,6 +570,14 @@ func mapJobInfo(ji *slurm.JobInfo) model.JobResponse {
|
|||||||
resp.Tasks = mapUint32NoValToInt32(ji.Tasks)
|
resp.Tasks = mapUint32NoValToInt32(ji.Tasks)
|
||||||
resp.NodeCount = mapUint32NoValToInt32(ji.NodeCount)
|
resp.NodeCount = mapUint32NoValToInt32(ji.NodeCount)
|
||||||
resp.BatchHost = derefStr(ji.BatchHost)
|
resp.BatchHost = derefStr(ji.BatchHost)
|
||||||
|
resp.CpusPerTask = mapUint16NoValToInt32(ji.CpusPerTask)
|
||||||
|
resp.MemoryPerCpu = mapUint64NoValToInt64(ji.MemoryPerCpu)
|
||||||
|
resp.MemoryPerNode = mapUint64NoValToInt64(ji.MemoryPerNode)
|
||||||
|
resp.TresReqStr = derefStr(ji.TresReqStr)
|
||||||
|
resp.TresAllocStr = derefStr(ji.TresAllocStr)
|
||||||
|
if ji.GresDetail != nil {
|
||||||
|
resp.GresDetail = []string(ji.GresDetail)
|
||||||
|
}
|
||||||
if ji.SubmitTime != nil && ji.SubmitTime.Number != nil {
|
if ji.SubmitTime != nil && ji.SubmitTime.Number != nil {
|
||||||
resp.SubmitTime = ji.SubmitTime.Number
|
resp.SubmitTime = ji.SubmitTime.Number
|
||||||
}
|
}
|
||||||
@@ -555,10 +648,22 @@ func mapSlurmdbJob(j *slurm.Job) model.JobResponse {
|
|||||||
}
|
}
|
||||||
if j.Required != nil {
|
if j.Required != nil {
|
||||||
resp.Cpus = j.Required.CPUs
|
resp.Cpus = j.Required.CPUs
|
||||||
|
resp.CpusPerTask = nil // SlurmDBD Job doesn't expose CpusPerTask directly
|
||||||
|
resp.MemoryPerCpu = mapUint64NoValToInt64(j.Required.MemoryPerCpu)
|
||||||
|
resp.MemoryPerNode = mapUint64NoValToInt64(j.Required.MemoryPerNode)
|
||||||
}
|
}
|
||||||
if j.AllocationNodes != nil {
|
if j.AllocationNodes != nil {
|
||||||
resp.NodeCount = j.AllocationNodes
|
resp.NodeCount = j.AllocationNodes
|
||||||
}
|
}
|
||||||
|
if j.Tres != nil {
|
||||||
|
resp.TresAllocated = formatTresList(j.Tres.Allocated)
|
||||||
|
resp.TresRequested = formatTresList(j.Tres.Requested)
|
||||||
|
}
|
||||||
|
if j.Time != nil {
|
||||||
|
resp.Elapsed = j.Time.Elapsed
|
||||||
|
}
|
||||||
|
resp.MemoryUsed = extractMemoryFromSteps(j.Steps)
|
||||||
|
resp.GresDetail = parseGresDetail(derefStr(j.UsedGres))
|
||||||
resp.WorkDir = derefStr(j.WorkingDirectory)
|
resp.WorkDir = derefStr(j.WorkingDirectory)
|
||||||
return resp
|
return resp
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -78,8 +78,8 @@ func TestSubmitJob_WithOptionalFields(t *testing.T) {
|
|||||||
if body.Job.Partition != nil {
|
if body.Job.Partition != nil {
|
||||||
t.Error("expected partition nil for empty string")
|
t.Error("expected partition nil for empty string")
|
||||||
}
|
}
|
||||||
if body.Job.MinimumCpus != nil {
|
if body.Job.CpusPerTask != nil {
|
||||||
t.Error("expected minimum_cpus nil when CPUs=0")
|
t.Error("expected cpus_per_task nil when CPUs=0")
|
||||||
}
|
}
|
||||||
|
|
||||||
jobID := int32(456)
|
jobID := int32(456)
|
||||||
@@ -885,18 +885,22 @@ func TestSubmitJob_AllSchedulingFields(t *testing.T) {
|
|||||||
if j.Name == nil || *j.Name != "full-test" {
|
if j.Name == nil || *j.Name != "full-test" {
|
||||||
t.Errorf("Name mismatch: %v", j.Name)
|
t.Errorf("Name mismatch: %v", j.Name)
|
||||||
}
|
}
|
||||||
if j.MinimumCpus == nil || *j.MinimumCpus != int32(8) {
|
// CPUs=8 maps to CpusPerTask, then overridden by explicit CpusPerTask=2
|
||||||
t.Errorf("MinimumCpus mismatch: %v", j.MinimumCpus)
|
if j.CpusPerTask == nil || *j.CpusPerTask != cpusPerTask {
|
||||||
|
t.Errorf("CpusPerTask mismatch: got %v, want %d (explicit CpusPerTask overrides CPUs)", j.CpusPerTask, cpusPerTask)
|
||||||
|
}
|
||||||
|
if j.MinimumCpus != nil {
|
||||||
|
t.Errorf("MinimumCpus should be nil, got %v", j.MinimumCpus)
|
||||||
}
|
}
|
||||||
|
|
||||||
// --- 22 new scheduling fields ---
|
// --- 22 new scheduling fields ---
|
||||||
|
|
||||||
// MemoryPerNode → *Uint64NoVal
|
// MemoryPerNode → *Uint64NoVal
|
||||||
if j.MemoryPerNode == nil || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
|
if j.MemoryPerNode == nil || j.MemoryPerNode.Set == nil || !*j.MemoryPerNode.Set || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
|
||||||
t.Errorf("MemoryPerNode mismatch: %v", j.MemoryPerNode)
|
t.Errorf("MemoryPerNode mismatch: %v", j.MemoryPerNode)
|
||||||
}
|
}
|
||||||
// MemoryPerCpu → *Uint64NoVal
|
// MemoryPerCpu → *Uint64NoVal
|
||||||
if j.MemoryPerCpu == nil || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
|
if j.MemoryPerCpu == nil || j.MemoryPerCpu.Set == nil || !*j.MemoryPerCpu.Set || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
|
||||||
t.Errorf("MemoryPerCpu mismatch: %v", j.MemoryPerCpu)
|
t.Errorf("MemoryPerCpu mismatch: %v", j.MemoryPerCpu)
|
||||||
}
|
}
|
||||||
// Nodes → *string
|
// Nodes → *string
|
||||||
@@ -1151,10 +1155,10 @@ func TestSubmitJob_MemoryBothSet(t *testing.T) {
|
|||||||
j := body.Job
|
j := body.Job
|
||||||
|
|
||||||
// Both memory fields should be mapped independently
|
// Both memory fields should be mapped independently
|
||||||
if j.MemoryPerNode == nil || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
|
if j.MemoryPerNode == nil || j.MemoryPerNode.Set == nil || !*j.MemoryPerNode.Set || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
|
||||||
t.Errorf("MemoryPerNode mismatch: %v", j.MemoryPerNode)
|
t.Errorf("MemoryPerNode mismatch: %v", j.MemoryPerNode)
|
||||||
}
|
}
|
||||||
if j.MemoryPerCpu == nil || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
|
if j.MemoryPerCpu == nil || j.MemoryPerCpu.Set == nil || !*j.MemoryPerCpu.Set || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
|
||||||
t.Errorf("MemoryPerCpu mismatch: %v", j.MemoryPerCpu)
|
t.Errorf("MemoryPerCpu mismatch: %v", j.MemoryPerCpu)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -123,3 +123,28 @@ func RandomSuffix(n int) string {
|
|||||||
}
|
}
|
||||||
return string(b)
|
return string(b)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func ResolveSchedulingMap(field string, task *model.Task) string {
|
||||||
|
switch field {
|
||||||
|
case "cpus":
|
||||||
|
return derefInt32ToStr(task.Cpus)
|
||||||
|
case "memory_per_node":
|
||||||
|
return derefInt64ToStr(task.MemoryPerNode)
|
||||||
|
case "memory_per_cpu":
|
||||||
|
return derefInt64ToStr(task.MemoryPerCpu)
|
||||||
|
case "nodes":
|
||||||
|
return derefStr(task.Nodes)
|
||||||
|
case "tasks":
|
||||||
|
return derefInt32ToStr(task.Tasks)
|
||||||
|
case "cpus_per_task":
|
||||||
|
return derefInt32ToStr(task.CpusPerTask)
|
||||||
|
case "partition":
|
||||||
|
return task.Partition
|
||||||
|
case "time_limit":
|
||||||
|
return derefInt32ToStr(task.TimeLimit)
|
||||||
|
case "qos":
|
||||||
|
return derefStr(task.QOS)
|
||||||
|
default:
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,72 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"gcy_hpc_server/internal/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
func strPtr(v string) *string { return &v }
|
||||||
|
|
||||||
|
func TestResolveSchedulingMap(t *testing.T) {
|
||||||
|
cpus := int32Ptr(8)
|
||||||
|
memPerNode := int64Ptr(4096)
|
||||||
|
memPerCpu := int64Ptr(512)
|
||||||
|
nodes := strPtr("2-4")
|
||||||
|
tasks := int32Ptr(4)
|
||||||
|
cpusPerTask := int32Ptr(2)
|
||||||
|
timeLimit := int32Ptr(60)
|
||||||
|
qos := strPtr("high")
|
||||||
|
|
||||||
|
task := &model.Task{
|
||||||
|
Partition: "gpu",
|
||||||
|
Cpus: cpus,
|
||||||
|
MemoryPerNode: memPerNode,
|
||||||
|
MemoryPerCpu: memPerCpu,
|
||||||
|
Nodes: nodes,
|
||||||
|
Tasks: tasks,
|
||||||
|
CpusPerTask: cpusPerTask,
|
||||||
|
TimeLimit: timeLimit,
|
||||||
|
QOS: qos,
|
||||||
|
}
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
field string
|
||||||
|
want string
|
||||||
|
}{
|
||||||
|
{"cpus", "8"},
|
||||||
|
{"memory_per_node", "4096"},
|
||||||
|
{"memory_per_cpu", "512"},
|
||||||
|
{"nodes", "2-4"},
|
||||||
|
{"tasks", "4"},
|
||||||
|
{"cpus_per_task", "2"},
|
||||||
|
{"partition", "gpu"},
|
||||||
|
{"time_limit", "60"},
|
||||||
|
{"qos", "high"},
|
||||||
|
{"unknown_field", ""},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.field, func(t *testing.T) {
|
||||||
|
got := ResolveSchedulingMap(tt.field, task)
|
||||||
|
if got != tt.want {
|
||||||
|
t.Errorf("ResolveSchedulingMap(%q) = %q, want %q", tt.field, got, tt.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestResolveSchedulingMap_NilFields(t *testing.T) {
|
||||||
|
// All scheduling fields are nil/empty — should return empty strings
|
||||||
|
task := &model.Task{}
|
||||||
|
for _, field := range []string{"cpus", "memory_per_node", "memory_per_cpu", "nodes", "tasks", "cpus_per_task", "time_limit", "qos"} {
|
||||||
|
got := ResolveSchedulingMap(field, task)
|
||||||
|
if got != "" {
|
||||||
|
t.Errorf("ResolveSchedulingMap(%q) with nil fields = %q, want empty", field, got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// partition is a plain string, not a pointer — empty string is the zero value
|
||||||
|
if got := ResolveSchedulingMap("partition", task); got != "" {
|
||||||
|
t.Errorf("ResolveSchedulingMap(partition) = %q, want empty", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -29,12 +29,29 @@ type TaskService struct {
|
|||||||
logger *zap.Logger
|
logger *zap.Logger
|
||||||
|
|
||||||
// async processing
|
// async processing
|
||||||
taskCh chan int64 // buffered channel, cap=16
|
taskCh chan int64 // buffered channel for task IDs awaiting processing
|
||||||
cancelFn context.CancelFunc
|
cancelFn context.CancelFunc
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
mu sync.Mutex // protects taskCh from send-on-closed
|
mu sync.Mutex // protects taskCh from send-on-closed
|
||||||
started bool // prevent double-start
|
started bool // prevent double-start
|
||||||
stopped bool
|
stopped bool
|
||||||
|
// inflight tracks task IDs currently being processed by the worker goroutine.
|
||||||
|
//
|
||||||
|
// Why it exists: taskCh is an in-memory Go channel — all pending taskIDs are
|
||||||
|
// lost when the server restarts. RecoverStuckTasks is responsible for
|
||||||
|
// recovering those lost tasks from the DB. However, GetStuckTasks uses a
|
||||||
|
// broad query (status NOT IN completed/failed AND updated_at < 5min ago) that
|
||||||
|
// also matches tasks being actively processed by the worker (e.g. a slow
|
||||||
|
// download). Without inflight, RecoverStuckTasks would reset those tasks to
|
||||||
|
// "submitted" and re-enqueue them, causing double-processing.
|
||||||
|
//
|
||||||
|
// How it works:
|
||||||
|
// - ProcessTask stores the taskID on entry, deletes on exit (via defer).
|
||||||
|
// - RecoverStuckTasks checks inflight before re-enqueueing; in-flight tasks
|
||||||
|
// are skipped.
|
||||||
|
// - On server restart inflight is empty (in-memory), so all genuinely stuck
|
||||||
|
// tasks are correctly recovered without false negatives.
|
||||||
|
inflight sync.Map // map[int64]struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewTaskService(
|
func NewTaskService(
|
||||||
@@ -56,7 +73,7 @@ func NewTaskService(
|
|||||||
jobSvc: jobSvc,
|
jobSvc: jobSvc,
|
||||||
workDirBase: workDirBase,
|
workDirBase: workDirBase,
|
||||||
logger: logger,
|
logger: logger,
|
||||||
taskCh: make(chan int64, 16),
|
taskCh: make(chan int64, 10000),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -169,6 +186,9 @@ func (s *TaskService) CreateTask(ctx context.Context, req *model.CreateTaskReque
|
|||||||
|
|
||||||
// ProcessTask runs the full synchronous processing pipeline for a task.
|
// ProcessTask runs the full synchronous processing pipeline for a task.
|
||||||
func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
|
func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
|
||||||
|
s.inflight.Store(taskID, struct{}{})
|
||||||
|
defer s.inflight.Delete(taskID)
|
||||||
|
|
||||||
// 1. Fetch task
|
// 1. Fetch task
|
||||||
task, err := s.taskStore.GetByID(ctx, taskID)
|
task, err := s.taskStore.GetByID(ctx, taskID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -178,6 +198,24 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
|
|||||||
return fmt.Errorf("task %d not found", taskID)
|
return fmt.Errorf("task %d not found", taskID)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Defense-in-depth against duplicate processing. When the same taskID enters
|
||||||
|
// taskCh multiple times (e.g. submitted normally + RecoverStuckTasks also
|
||||||
|
// enqueues it before the worker picks up the first copy), the worker processes
|
||||||
|
// them sequentially. The first invocation changes status from "submitted" to
|
||||||
|
// "preparing"; the second invocation reads the latest DB status, sees
|
||||||
|
// non-submitted, and safely skips.
|
||||||
|
//
|
||||||
|
// This does NOT block retries: processWithRetry sets status back to "submitted"
|
||||||
|
// before re-enqueueing, so the retried invocation passes this check and
|
||||||
|
// continues from the saved currentStep.
|
||||||
|
if task.Status != model.TaskStatusSubmitted {
|
||||||
|
s.logger.Debug("skipping task with non-submitted status",
|
||||||
|
zap.Int64("task_id", taskID),
|
||||||
|
zap.String("status", string(task.Status)),
|
||||||
|
)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
fail := func(step, msg string) error {
|
fail := func(step, msg string) error {
|
||||||
_ = s.taskStore.UpdateStatus(ctx, taskID, model.TaskStatusFailed, msg)
|
_ = s.taskStore.UpdateStatus(ctx, taskID, model.TaskStatusFailed, msg)
|
||||||
_ = s.taskStore.UpdateRetryState(ctx, taskID, model.TaskStatusFailed, step, task.RetryCount)
|
_ = s.taskStore.UpdateRetryState(ctx, taskID, model.TaskStatusFailed, step, task.RetryCount)
|
||||||
@@ -263,7 +301,15 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 13-14. Set ready + submitting
|
// 13-14. Set ready + submitting (guard: skip if already submitted to Slurm)
|
||||||
|
if task.SlurmJobID != nil {
|
||||||
|
s.logger.Info("task already has slurm job, skipping submission",
|
||||||
|
zap.Int64("task_id", taskID),
|
||||||
|
zap.Int32("slurm_job_id", *task.SlurmJobID),
|
||||||
|
)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
if err := s.taskStore.UpdateRetryState(ctx, taskID, model.TaskStatusReady, model.TaskStepSubmitting, 0); err != nil {
|
if err := s.taskStore.UpdateRetryState(ctx, taskID, model.TaskStatusReady, model.TaskStepSubmitting, 0); err != nil {
|
||||||
return fail(model.TaskStepSubmitting, fmt.Sprintf("update status to ready: %v", err))
|
return fail(model.TaskStepSubmitting, fmt.Sprintf("update status to ready: %v", err))
|
||||||
}
|
}
|
||||||
@@ -276,6 +322,14 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 15a. Parse app environment
|
||||||
|
var appEnv map[string]string
|
||||||
|
if len(app.Environment) > 0 {
|
||||||
|
if err := json.Unmarshal(app.Environment, &appEnv); err != nil {
|
||||||
|
return fail(model.TaskStepSubmitting, fmt.Sprintf("parse application environment: %v", err))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// 16. Parse task values
|
// 16. Parse task values
|
||||||
values := make(map[string]string)
|
values := make(map[string]string)
|
||||||
if len(task.Values) > 0 {
|
if len(task.Values) > 0 {
|
||||||
@@ -284,16 +338,55 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 16a. Auto-inject WORK_DIR if the app defines it as a parameter.
|
||||||
|
// The work directory is created by the server, not provided by the user.
|
||||||
|
for _, p := range params {
|
||||||
|
if p.Name == "WORK_DIR" {
|
||||||
|
values["WORK_DIR"] = workDir
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 16b. Map input_file_ids to file-type parameters by order.
|
||||||
|
// User selects files via FilePicker; we assign their IDs to file/directory
|
||||||
|
// params sequentially so the backend can resolve them to filenames later.
|
||||||
|
var inputFileIDs []int64
|
||||||
|
if len(task.InputFileIDs) > 0 {
|
||||||
|
if err := json.Unmarshal(task.InputFileIDs, &inputFileIDs); err != nil {
|
||||||
|
return fail(model.TaskStepSubmitting, fmt.Sprintf("parse input file ids: %v", err))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(inputFileIDs) > 0 {
|
||||||
|
fileParamIdx := 0
|
||||||
|
for _, p := range params {
|
||||||
|
if p.Type != model.ParamTypeFile && p.Type != model.ParamTypeDirectory {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if fileParamIdx < len(inputFileIDs) {
|
||||||
|
values[p.Name] = strconv.FormatInt(inputFileIDs[fileParamIdx], 10)
|
||||||
|
fileParamIdx++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 16b-3. Auto-inject scheduling params based on scheduling_map.
|
||||||
|
// If an Application parameter declares scheduling_map, the corresponding
|
||||||
|
// scheduling field value overrides any user-provided value.
|
||||||
|
for _, p := range params {
|
||||||
|
if p.SchedulingMap == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if val := ResolveSchedulingMap(p.SchedulingMap, task); val != "" {
|
||||||
|
values[p.Name] = val
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 16c. Validate all params (WORK_DIR and file params now have values).
|
||||||
if err := ValidateParams(params, values); err != nil {
|
if err := ValidateParams(params, values); err != nil {
|
||||||
return fail(model.TaskStepSubmitting, err.Error())
|
return fail(model.TaskStepSubmitting, err.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
if strings.Contains(app.ScriptTemplate, "$WORK_DIR") {
|
// 16d. Resolve file-type parameter values: file_id → filename.
|
||||||
values["WORK_DIR"] = workDir
|
|
||||||
}
|
|
||||||
|
|
||||||
// Resolve file-type parameters: user sends file_id, we replace with filename.
|
|
||||||
// Only query the database if there are file/directory-type parameters with values.
|
|
||||||
var fileLookupIDs []int64
|
var fileLookupIDs []int64
|
||||||
for _, p := range params {
|
for _, p := range params {
|
||||||
if p.Type != model.ParamTypeFile && p.Type != model.ParamTypeDirectory {
|
if p.Type != model.ParamTypeFile && p.Type != model.ParamTypeDirectory {
|
||||||
@@ -386,6 +479,7 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
|
|||||||
Dependency: task.Dependency,
|
Dependency: task.Dependency,
|
||||||
Requeue: task.Requeue,
|
Requeue: task.Requeue,
|
||||||
KillOnNodeFail: task.KillOnNodeFail,
|
KillOnNodeFail: task.KillOnNodeFail,
|
||||||
|
Environment: appEnv,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fail(model.TaskStepSubmitting, fmt.Sprintf("submit job: %v", err))
|
return fail(model.TaskStepSubmitting, fmt.Sprintf("submit job: %v", err))
|
||||||
@@ -453,7 +547,7 @@ func uniqueInt64s(ids []int64) []int64 {
|
|||||||
|
|
||||||
func (s *TaskService) mapSlurmStateToTaskStatus(slurmState []string) string {
|
func (s *TaskService) mapSlurmStateToTaskStatus(slurmState []string) string {
|
||||||
if len(slurmState) == 0 {
|
if len(slurmState) == 0 {
|
||||||
return model.TaskStatusRunning
|
return ""
|
||||||
}
|
}
|
||||||
|
|
||||||
state := strings.ToUpper(slurmState[0])
|
state := strings.ToUpper(slurmState[0])
|
||||||
@@ -467,7 +561,8 @@ func (s *TaskService) mapSlurmStateToTaskStatus(slurmState []string) string {
|
|||||||
case "FAILED", "CANCELLED", "TIMEOUT", "NODE_FAIL", "OUT_OF_MEMORY", "PREEMPTED":
|
case "FAILED", "CANCELLED", "TIMEOUT", "NODE_FAIL", "OUT_OF_MEMORY", "PREEMPTED":
|
||||||
return model.TaskStatusFailed
|
return model.TaskStatusFailed
|
||||||
default:
|
default:
|
||||||
return model.TaskStatusRunning
|
s.logger.Warn("unrecognized slurm state, skipping update", zap.String("state", state))
|
||||||
|
return ""
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -498,15 +593,16 @@ func (s *TaskService) refreshTaskStatus(ctx context.Context, taskID int64) error
|
|||||||
}
|
}
|
||||||
|
|
||||||
newStatus := s.mapSlurmStateToTaskStatus(jobResp.State)
|
newStatus := s.mapSlurmStateToTaskStatus(jobResp.State)
|
||||||
if newStatus != task.Status {
|
if newStatus == "" || newStatus == task.Status {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
s.logger.Info("updating task status from slurm",
|
s.logger.Info("updating task status from slurm",
|
||||||
zap.Int64("task_id", taskID),
|
zap.Int64("task_id", taskID),
|
||||||
zap.String("old_status", task.Status),
|
zap.String("old_status", task.Status),
|
||||||
zap.String("new_status", newStatus),
|
zap.String("new_status", newStatus),
|
||||||
)
|
)
|
||||||
return s.taskStore.UpdateStatus(ctx, taskID, newStatus, "")
|
return s.taskStore.UpdateStatus(ctx, taskID, newStatus, "")
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *TaskService) RefreshStaleTasks(ctx context.Context) error {
|
func (s *TaskService) RefreshStaleTasks(ctx context.Context) error {
|
||||||
@@ -618,7 +714,7 @@ func (s *TaskService) StopProcessor() {
|
|||||||
|
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
drainCh := s.taskCh
|
drainCh := s.taskCh
|
||||||
s.taskCh = make(chan int64, 16)
|
s.taskCh = make(chan int64, 10000)
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
|
|
||||||
for taskID := range drainCh {
|
for taskID := range drainCh {
|
||||||
@@ -652,12 +748,55 @@ func (s *TaskService) processWithRetry(ctx context.Context, taskID int64) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *TaskService) RecoverStuckTasks(ctx context.Context) {
|
func (s *TaskService) RecoverStuckTasks(ctx context.Context) {
|
||||||
|
// RecoverStuckTasks recovers tasks that are "stuck" — they exist in the DB
|
||||||
|
// with a non-terminal status but are not being processed.
|
||||||
|
//
|
||||||
|
// Scenarios that create stuck tasks:
|
||||||
|
//
|
||||||
|
// 1. Server restart: taskCh is an in-memory Go channel, all pending IDs are
|
||||||
|
// lost on process exit. Tasks that were queued but never picked up by the
|
||||||
|
// worker remain in "submitted" status in DB with no one to process them.
|
||||||
|
//
|
||||||
|
// 2. Server crash mid-processing: the worker had advanced a task to
|
||||||
|
// "preparing"/"downloading" and then died. The task sits in that
|
||||||
|
// intermediate state with no SlurmJobID and no worker to continue.
|
||||||
|
//
|
||||||
|
// 3. Channel full: SubmitAsync dropped a task because taskCh was at
|
||||||
|
// capacity. The task stays "submitted" but was never enqueued.
|
||||||
|
//
|
||||||
|
// The bug this fix addresses:
|
||||||
|
//
|
||||||
|
// GetStuckTasks queries: status NOT IN (completed, failed) AND updated_at <
|
||||||
|
// 5min ago. This also matches tasks currently being processed by the worker
|
||||||
|
// whose step is slow (>5 min, e.g. downloading large files) and hasn't
|
||||||
|
// refreshed updated_at. Without the inflight check below, this function
|
||||||
|
// would reset such a task to "submitted" and re-enqueue it, causing the
|
||||||
|
// same task to be processed by two concurrent invocations of ProcessTask.
|
||||||
|
//
|
||||||
|
// Fix: the inflight sync.Map tracks taskIDs currently inside ProcessTask.
|
||||||
|
// Tasks found in inflight are skipped here. On server restart inflight is
|
||||||
|
// empty (it's in-memory), so all genuinely stuck tasks from scenarios 1-3
|
||||||
|
// above are correctly recovered.
|
||||||
|
|
||||||
tasks, err := s.taskStore.GetStuckTasks(ctx, 5*time.Minute)
|
tasks, err := s.taskStore.GetStuckTasks(ctx, 5*time.Minute)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Error("failed to get stuck tasks", zap.Error(err))
|
s.logger.Error("failed to get stuck tasks", zap.Error(err))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
for i := range tasks {
|
for i := range tasks {
|
||||||
|
if tasks[i].SlurmJobID != nil {
|
||||||
|
s.logger.Info("skipping stuck task recovery, already in slurm",
|
||||||
|
zap.Int64("taskID", tasks[i].ID),
|
||||||
|
zap.Int32("slurm_job_id", *tasks[i].SlurmJobID),
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, ok := s.inflight.Load(tasks[i].ID); ok {
|
||||||
|
s.logger.Debug("skipping in-flight task",
|
||||||
|
zap.Int64("taskID", tasks[i].ID),
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
}
|
||||||
_ = s.taskStore.UpdateStatus(ctx, tasks[i].ID, model.TaskStatusSubmitted, "")
|
_ = s.taskStore.UpdateStatus(ctx, tasks[i].ID, model.TaskStatusSubmitted, "")
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
if !s.stopped {
|
if !s.stopped {
|
||||||
|
|||||||
@@ -97,7 +97,7 @@ func TestTaskService_MapSlurmState_AllStates(t *testing.T) {
|
|||||||
{[]string{"OUT_OF_MEMORY"}, model.TaskStatusFailed},
|
{[]string{"OUT_OF_MEMORY"}, model.TaskStatusFailed},
|
||||||
{[]string{"PREEMPTED"}, model.TaskStatusFailed},
|
{[]string{"PREEMPTED"}, model.TaskStatusFailed},
|
||||||
{[]string{"SPECIAL_EXIT"}, model.TaskStatusRunning},
|
{[]string{"SPECIAL_EXIT"}, model.TaskStatusRunning},
|
||||||
{[]string{"unknown_state"}, model.TaskStatusRunning},
|
{[]string{"unknown_state"}, ""},
|
||||||
{[]string{"pending"}, model.TaskStatusQueued},
|
{[]string{"pending"}, model.TaskStatusQueued},
|
||||||
{[]string{"Running"}, model.TaskStatusRunning},
|
{[]string{"Running"}, model.TaskStatusRunning},
|
||||||
}
|
}
|
||||||
@@ -115,13 +115,13 @@ func TestTaskService_MapSlurmState_Empty(t *testing.T) {
|
|||||||
defer env.close()
|
defer env.close()
|
||||||
|
|
||||||
got := env.svc.mapSlurmStateToTaskStatus([]string{})
|
got := env.svc.mapSlurmStateToTaskStatus([]string{})
|
||||||
if got != model.TaskStatusRunning {
|
if got != "" {
|
||||||
t.Errorf("mapSlurmStateToTaskStatus([]) = %q, want %q", got, model.TaskStatusRunning)
|
t.Errorf("mapSlurmStateToTaskStatus([]) = %q, want empty string", got)
|
||||||
}
|
}
|
||||||
|
|
||||||
got = env.svc.mapSlurmStateToTaskStatus(nil)
|
got = env.svc.mapSlurmStateToTaskStatus(nil)
|
||||||
if got != model.TaskStatusRunning {
|
if got != "" {
|
||||||
t.Errorf("mapSlurmStateToTaskStatus(nil) = %q, want %q", got, model.TaskStatusRunning)
|
t.Errorf("mapSlurmStateToTaskStatus(nil) = %q, want empty string", got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1048,8 +1048,12 @@ func TestProcessTask_SchedulingParams(t *testing.T) {
|
|||||||
if j.Partition == nil || *j.Partition != "gpu" {
|
if j.Partition == nil || *j.Partition != "gpu" {
|
||||||
t.Errorf("Partition = %v, want %q", j.Partition, "gpu")
|
t.Errorf("Partition = %v, want %q", j.Partition, "gpu")
|
||||||
}
|
}
|
||||||
if j.MinimumCpus == nil || *j.MinimumCpus != int32(8) {
|
// CPUs=8 maps to CpusPerTask, then overridden by explicit CpusPerTask=2
|
||||||
t.Errorf("MinimumCpus = %v, want 8", j.MinimumCpus)
|
if j.CpusPerTask == nil || *j.CpusPerTask != int32(2) {
|
||||||
|
t.Errorf("CpusPerTask = %v, want 2 (explicit CpusPerTask overrides CPUs)", j.CpusPerTask)
|
||||||
|
}
|
||||||
|
if j.MinimumCpus != nil {
|
||||||
|
t.Errorf("MinimumCpus should be nil, got %v", j.MinimumCpus)
|
||||||
}
|
}
|
||||||
if j.TimeLimit == nil || j.TimeLimit.Number == nil || *j.TimeLimit.Number != int64(60) {
|
if j.TimeLimit == nil || j.TimeLimit.Number == nil || *j.TimeLimit.Number != int64(60) {
|
||||||
t.Errorf("TimeLimit = %v, want 60", j.TimeLimit)
|
t.Errorf("TimeLimit = %v, want 60", j.TimeLimit)
|
||||||
@@ -1175,8 +1179,8 @@ func TestProcessTask_PartialSchedulingParams(t *testing.T) {
|
|||||||
t.Errorf("Partition = %v, want %q", j.Partition, "debug")
|
t.Errorf("Partition = %v, want %q", j.Partition, "debug")
|
||||||
}
|
}
|
||||||
|
|
||||||
if j.MinimumCpus != nil {
|
if j.CpusPerTask != nil {
|
||||||
t.Errorf("MinimumCpus = %v, want nil (no cpus set)", j.MinimumCpus)
|
t.Errorf("CpusPerTask = %v, want nil (no cpus set)", j.CpusPerTask)
|
||||||
}
|
}
|
||||||
if j.TimeLimit == nil {
|
if j.TimeLimit == nil {
|
||||||
t.Errorf("TimeLimit = nil, want non-nil (default should be injected)")
|
t.Errorf("TimeLimit = nil, want non-nil (default should be injected)")
|
||||||
@@ -1262,3 +1266,49 @@ func TestProcessTask_PartialSchedulingParams(t *testing.T) {
|
|||||||
t.Errorf("KillOnNodeFail = %v, want nil", j.KillOnNodeFail)
|
t.Errorf("KillOnNodeFail = %v, want nil", j.KillOnNodeFail)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTaskService_ProcessTask_SchedulingMapInjection(t *testing.T) {
|
||||||
|
jobID := int32(42)
|
||||||
|
|
||||||
|
var capturedReq slurm.JobSubmitReq
|
||||||
|
|
||||||
|
env := newTaskTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if err := json.NewDecoder(r.Body).Decode(&capturedReq); err != nil {
|
||||||
|
t.Fatalf("decode request body: %v", err)
|
||||||
|
}
|
||||||
|
json.NewEncoder(w).Encode(slurm.OpenapiJobSubmitResponse{
|
||||||
|
Result: &slurm.JobSubmitResponseMsg{JobID: &jobID},
|
||||||
|
})
|
||||||
|
}))
|
||||||
|
defer env.close()
|
||||||
|
|
||||||
|
params := json.RawMessage(`[
|
||||||
|
{"name": "NP", "type": "integer", "scheduling_map": "cpus", "required": true}
|
||||||
|
]`)
|
||||||
|
appID := env.createApp(t, "sched-map-app", "#!/bin/bash\nmpirun -np $NP my_app", params)
|
||||||
|
|
||||||
|
cpus := int32(8)
|
||||||
|
task, err := env.svc.CreateTask(context.Background(), &model.CreateTaskRequest{
|
||||||
|
AppID: appID,
|
||||||
|
TaskName: "sched-map-test",
|
||||||
|
Cpus: &cpus,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("CreateTask: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := env.svc.ProcessTask(context.Background(), task.ID); err != nil {
|
||||||
|
t.Fatalf("ProcessTask: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if capturedReq.Script == nil {
|
||||||
|
t.Fatal("submitted script is nil")
|
||||||
|
}
|
||||||
|
|
||||||
|
if !strings.Contains(*capturedReq.Script, "'8'") {
|
||||||
|
t.Errorf("rendered script does not contain shell-escaped scheduling value:\n%s", *capturedReq.Script)
|
||||||
|
}
|
||||||
|
if !strings.Contains(*capturedReq.Script, "mpirun -np '8'") {
|
||||||
|
t.Errorf("rendered script does not contain expected mpirun command:\n%s", *capturedReq.Script)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -357,8 +357,9 @@ func (s *UploadService) CompleteUpload(ctx context.Context, sessionID int64) (*m
|
|||||||
keys[i] = fmt.Sprintf("%schunk_%05d", minioPrefix, i)
|
keys[i] = fmt.Sprintf("%schunk_%05d", minioPrefix, i)
|
||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
bgCtx := context.Background()
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||||
if delErr := s.storage.RemoveObjects(bgCtx, s.cfg.Bucket, keys, storage.RemoveObjectsOptions{}); delErr != nil {
|
defer cancel()
|
||||||
|
if delErr := s.storage.RemoveObjects(ctx, s.cfg.Bucket, keys, storage.RemoveObjectsOptions{}); delErr != nil {
|
||||||
s.logger.Warn("delete temp chunks", zap.Error(delErr))
|
s.logger.Warn("delete temp chunks", zap.Error(delErr))
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
"gcy_hpc_server/internal/model"
|
"gcy_hpc_server/internal/model"
|
||||||
|
|
||||||
@@ -52,6 +53,15 @@ func (s *ApplicationStore) Create(ctx context.Context, req *model.CreateApplicat
|
|||||||
params = json.RawMessage(`[]`)
|
params = json.RawMessage(`[]`)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var envJSON json.RawMessage
|
||||||
|
if len(req.Environment) > 0 {
|
||||||
|
b, err := json.Marshal(req.Environment)
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("marshal environment: %w", err)
|
||||||
|
}
|
||||||
|
envJSON = b
|
||||||
|
}
|
||||||
|
|
||||||
app := &model.Application{
|
app := &model.Application{
|
||||||
Name: req.Name,
|
Name: req.Name,
|
||||||
Description: req.Description,
|
Description: req.Description,
|
||||||
@@ -59,6 +69,7 @@ func (s *ApplicationStore) Create(ctx context.Context, req *model.CreateApplicat
|
|||||||
Category: req.Category,
|
Category: req.Category,
|
||||||
ScriptTemplate: req.ScriptTemplate,
|
ScriptTemplate: req.ScriptTemplate,
|
||||||
Parameters: params,
|
Parameters: params,
|
||||||
|
Environment: envJSON,
|
||||||
Scope: req.Scope,
|
Scope: req.Scope,
|
||||||
}
|
}
|
||||||
if err := s.db.WithContext(ctx).Create(app).Error; err != nil {
|
if err := s.db.WithContext(ctx).Create(app).Error; err != nil {
|
||||||
@@ -87,6 +98,9 @@ func (s *ApplicationStore) Update(ctx context.Context, id int64, req *model.Upda
|
|||||||
if req.Parameters != nil {
|
if req.Parameters != nil {
|
||||||
updates["parameters"] = *req.Parameters
|
updates["parameters"] = *req.Parameters
|
||||||
}
|
}
|
||||||
|
if req.Environment != nil {
|
||||||
|
updates["environment"] = *req.Environment
|
||||||
|
}
|
||||||
if req.Scope != nil {
|
if req.Scope != nil {
|
||||||
updates["scope"] = *req.Scope
|
updates["scope"] = *req.Scope
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -49,7 +49,7 @@ func TestAllRoutesRegistered(t *testing.T) {
|
|||||||
method string
|
method string
|
||||||
path string
|
path string
|
||||||
}{
|
}{
|
||||||
{"POST", "/api/v1/jobs/submit"},
|
// {"POST", "/api/v1/jobs/submit"}, // [已弃用] 已被 POST /tasks 取代
|
||||||
{"GET", "/api/v1/jobs"},
|
{"GET", "/api/v1/jobs"},
|
||||||
{"GET", "/api/v1/jobs/history"},
|
{"GET", "/api/v1/jobs/history"},
|
||||||
{"GET", "/api/v1/jobs/1"},
|
{"GET", "/api/v1/jobs/1"},
|
||||||
@@ -82,8 +82,8 @@ func TestAllRoutesRegistered(t *testing.T) {
|
|||||||
{"GET", "/api/v1/tasks"},
|
{"GET", "/api/v1/tasks"},
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(routes) != 30 {
|
if len(routes) != 29 {
|
||||||
t.Fatalf("expected 31 routes, got %d", len(routes))
|
t.Fatalf("expected 30 routes, got %d", len(routes))
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, r := range routes {
|
for _, r := range routes {
|
||||||
|
|||||||
Generated
+1621
-1085
File diff suppressed because it is too large
Load Diff
+14
-13
@@ -9,20 +9,21 @@
|
|||||||
"preview": "vite preview"
|
"preview": "vite preview"
|
||||||
},
|
},
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"@element-plus/icons-vue": "^2.3.2",
|
"@element-plus/icons-vue": "^2.3.1",
|
||||||
"axios": "^1.15.0",
|
"axios": "^1.6.5",
|
||||||
"element-plus": "^2.13.7",
|
"element-plus": "^2.5.2",
|
||||||
"vue": "^3.5.32",
|
"vue": "^3.4.15",
|
||||||
"vue-router": "^5.0.4"
|
"vue-router": "^4.2.5"
|
||||||
},
|
},
|
||||||
"devDependencies": {
|
"devDependencies": {
|
||||||
"@types/node": "^24.12.2",
|
"@playwright/test": "^1.59.1",
|
||||||
"@vitejs/plugin-vue": "^6.0.5",
|
"@types/node": "^20.11.5",
|
||||||
"@vue/tsconfig": "^0.9.1",
|
"@vitejs/plugin-vue": "^5.0.3",
|
||||||
"typescript": "~6.0.2",
|
"@vue/tsconfig": "^0.5.1",
|
||||||
"unplugin-auto-import": "^21.0.0",
|
"typescript": "~5.3.3",
|
||||||
"unplugin-vue-components": "^32.0.0",
|
"unplugin-auto-import": "^0.17.3",
|
||||||
"vite": "^8.0.4",
|
"unplugin-vue-components": "^0.26.0",
|
||||||
"vue-tsc": "^3.2.6"
|
"vite": "^5.0.12",
|
||||||
|
"vue-tsc": "^2.0.24"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ export interface ParameterSchema {
|
|||||||
default?: string
|
default?: string
|
||||||
options?: string[]
|
options?: string[]
|
||||||
description?: string
|
description?: string
|
||||||
|
scheduling_map?: string
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface Application {
|
export interface Application {
|
||||||
|
|||||||
@@ -34,7 +34,7 @@
|
|||||||
|
|
||||||
<template v-if="selectedApp">
|
<template v-if="selectedApp">
|
||||||
<el-form-item
|
<el-form-item
|
||||||
v-for="param in (selectedApp.parameters || [])"
|
v-for="param in visibleParams"
|
||||||
:key="param.name"
|
:key="param.name"
|
||||||
:label="param.label || param.name"
|
:label="param.label || param.name"
|
||||||
>
|
>
|
||||||
@@ -57,7 +57,13 @@
|
|||||||
@update:model-value="(val: string | number | boolean) => { values[param.name] = String(val) }"
|
@update:model-value="(val: string | number | boolean) => { values[param.name] = String(val) }"
|
||||||
/>
|
/>
|
||||||
|
|
||||||
<el-input v-else-if="param.type === 'file' || param.type === 'directory'" disabled placeholder="文件选择功能开发中" />
|
<template v-else-if="param.type === 'file' || param.type === 'directory'">
|
||||||
|
<div v-if="getFileForParam(param)" style="display: flex; align-items: center; gap: 8px">
|
||||||
|
<el-tag>{{ getFileForParam(param).name }}</el-tag>
|
||||||
|
<span style="color: #909399; font-size: 12px">ID: {{ getFileForParam(param).id }}</span>
|
||||||
|
</div>
|
||||||
|
<span v-else style="color: #909399; font-size: 13px">请通过上方"关联文件"选择</span>
|
||||||
|
</template>
|
||||||
</el-form-item>
|
</el-form-item>
|
||||||
</template>
|
</template>
|
||||||
|
|
||||||
@@ -115,6 +121,33 @@ const showFilePicker = ref(false)
|
|||||||
|
|
||||||
const selectedApp = computed(() => appList.value.find(a => a.id === selectedAppId.value))
|
const selectedApp = computed(() => appList.value.find(a => a.id === selectedAppId.value))
|
||||||
|
|
||||||
|
const autoParams = new Set(['WORK_DIR'])
|
||||||
|
|
||||||
|
const visibleParams = computed(() =>
|
||||||
|
(selectedApp.value?.parameters || []).filter((p: any) => !autoParams.has(p.name) && !p.scheduling_map)
|
||||||
|
)
|
||||||
|
|
||||||
|
const fileParams = computed(() =>
|
||||||
|
(selectedApp.value?.parameters || []).filter(
|
||||||
|
(p: any) => p.type === 'file' || p.type === 'directory'
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
const fileParamMapping = computed(() => {
|
||||||
|
const mapping: Record<string, string> = {}
|
||||||
|
fileParams.value.forEach((p: any, i: number) => {
|
||||||
|
if (selectedFiles.value[i]) {
|
||||||
|
mapping[p.name] = String(selectedFiles.value[i].id)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
return mapping
|
||||||
|
})
|
||||||
|
|
||||||
|
const getFileForParam = (param: any) => {
|
||||||
|
const index = fileParams.value.findIndex((p: any) => p.name === param.name)
|
||||||
|
return selectedFiles.value[index]
|
||||||
|
}
|
||||||
|
|
||||||
watch(selectedAppId, () => { values.value = {} })
|
watch(selectedAppId, () => { values.value = {} })
|
||||||
|
|
||||||
onMounted(async () => {
|
onMounted(async () => {
|
||||||
@@ -126,12 +159,33 @@ onMounted(async () => {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
const resolveSchedMapValue = (mapField: string): string | undefined => {
|
||||||
|
switch (mapField) {
|
||||||
|
case 'cpus': return form.cpus != null ? String(form.cpus) : undefined
|
||||||
|
case 'memory_per_node': return form.memory_per_node != null ? String(form.memory_per_node) : undefined
|
||||||
|
case 'nodes': return form.nodes || undefined
|
||||||
|
case 'tasks': return form.tasks != null ? String(form.tasks) : undefined
|
||||||
|
case 'cpus_per_task': return form.cpus_per_task != null ? String(form.cpus_per_task) : undefined
|
||||||
|
case 'partition': return form.partition || undefined
|
||||||
|
default: return undefined
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
const handleSubmit = async () => {
|
const handleSubmit = async () => {
|
||||||
if (!selectedAppId.value) { ElMessage.warning('请选择应用'); return }
|
if (!selectedAppId.value) { ElMessage.warning('请选择应用'); return }
|
||||||
submitting.value = true
|
submitting.value = true
|
||||||
try {
|
try {
|
||||||
const taskName = form.task_name.trim() || `task_${selectedAppId.value}_${Date.now()}`
|
const taskName = form.task_name.trim() || `task_${selectedAppId.value}_${Date.now()}`
|
||||||
const resp = await createTask({ ...form, task_name: taskName, job_name: taskName, app_id: selectedAppId.value, values: values.value, file_ids: selectedFiles.value.map(f => f.id) })
|
const mergedValues = { ...values.value, ...fileParamMapping.value }
|
||||||
|
// Auto-inject scheduling_map values
|
||||||
|
const schedParams = (selectedApp.value?.parameters || []).filter((p: any) => p.scheduling_map)
|
||||||
|
for (const p of schedParams) {
|
||||||
|
const val = resolveSchedMapValue(p.scheduling_map)
|
||||||
|
if (val !== undefined && val !== '') {
|
||||||
|
mergedValues[p.name] = val
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const resp = await createTask({ ...form, task_name: taskName, job_name: taskName, app_id: selectedAppId.value, values: mergedValues, file_ids: selectedFiles.value.map(f => f.id) })
|
||||||
if (resp.success) {
|
if (resp.success) {
|
||||||
ElMessage.success('任务提交成功')
|
ElMessage.success('任务提交成功')
|
||||||
router.push('/tasks')
|
router.push('/tasks')
|
||||||
|
|||||||
@@ -1,18 +1,17 @@
|
|||||||
{
|
{
|
||||||
"extends": "@vue/tsconfig/tsconfig.dom.json",
|
"extends": "@vue/tsconfig/tsconfig.dom.json",
|
||||||
"compilerOptions": {
|
"compilerOptions": {
|
||||||
|
"incremental": true,
|
||||||
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.app.tsbuildinfo",
|
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.app.tsbuildinfo",
|
||||||
"types": ["vite/client"],
|
"types": ["vite/client"],
|
||||||
"baseUrl": ".",
|
"baseUrl": ".",
|
||||||
"paths": {
|
"paths": {
|
||||||
"@/*": ["src/*"]
|
"@/*": ["src/*"]
|
||||||
},
|
},
|
||||||
"ignoreDeprecations": "6.0",
|
|
||||||
|
|
||||||
/* Linting */
|
/* Linting */
|
||||||
"noUnusedLocals": true,
|
"noUnusedLocals": true,
|
||||||
"noUnusedParameters": true,
|
"noUnusedParameters": true,
|
||||||
"erasableSyntaxOnly": true,
|
|
||||||
"noFallthroughCasesInSwitch": true
|
"noFallthroughCasesInSwitch": true
|
||||||
},
|
},
|
||||||
"include": ["src/**/*.ts", "src/**/*.tsx", "src/**/*.vue"]
|
"include": ["src/**/*.ts", "src/**/*.tsx", "src/**/*.vue"]
|
||||||
|
|||||||
@@ -1,8 +1,9 @@
|
|||||||
{
|
{
|
||||||
"compilerOptions": {
|
"compilerOptions": {
|
||||||
|
"composite": true,
|
||||||
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.node.tsbuildinfo",
|
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.node.tsbuildinfo",
|
||||||
"target": "es2023",
|
"target": "es2022",
|
||||||
"lib": ["ES2023"],
|
"lib": ["ES2022"],
|
||||||
"module": "esnext",
|
"module": "esnext",
|
||||||
"types": ["node"],
|
"types": ["node"],
|
||||||
"skipLibCheck": true,
|
"skipLibCheck": true,
|
||||||
@@ -10,14 +11,12 @@
|
|||||||
/* Bundler mode */
|
/* Bundler mode */
|
||||||
"moduleResolution": "bundler",
|
"moduleResolution": "bundler",
|
||||||
"allowImportingTsExtensions": true,
|
"allowImportingTsExtensions": true,
|
||||||
"verbatimModuleSyntax": true,
|
|
||||||
"moduleDetection": "force",
|
"moduleDetection": "force",
|
||||||
"noEmit": true,
|
"noEmit": true,
|
||||||
|
|
||||||
/* Linting */
|
/* Linting */
|
||||||
"noUnusedLocals": true,
|
"noUnusedLocals": true,
|
||||||
"noUnusedParameters": true,
|
"noUnusedParameters": true,
|
||||||
"erasableSyntaxOnly": true,
|
|
||||||
"noFallthroughCasesInSwitch": true
|
"noFallthroughCasesInSwitch": true
|
||||||
},
|
},
|
||||||
"include": ["vite.config.ts"]
|
"include": ["vite.config.ts"]
|
||||||
|
|||||||
Reference in New Issue
Block a user