diff --git a/Dockerfile b/Dockerfile index 87c6747..7065dcf 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,9 +1,45 @@ #FROM 172.20.32.187/pipeline-service/pipeline-convert:running -FROM ccr.ccs.tencentyun.com/somunslotus/pipeline-covert:running -WORKDIR /app -RUN go build -COPY ./remote-task-excutor-cli /app/remote-task-excutor-cli -#COPY ./internal/config/config.yaml /usr/local/argo-pipeline-manager/ -RUN chmod a+x /app/remote-task-excutor-cli -ENTRYPOINT ["/app/remote-task-excutor-cli"] +##FROM ccr.ccs.tencentyun.com/somunslotus/pipeline-covert:running +#WORKDIR /app +#RUN go build -o remote-task-excutor-cli +#COPY ./remote-task-excutor-cli /app/remote-task-excutor-cli +##COPY ./internal/config/config.yaml /usr/local/argo-pipeline-manager/ +#RUN chmod a+x /app/remote-task-excutor-cli +#ENTRYPOINT ["/app/remote-task-excutor-cli", "run"] + +# 第一阶段：构建阶段 +FROM 172.20.32.187/pipeline-service/golang:1.24.5-alpine3.22 AS builder + +WORKDIR /app + +# 先复制依赖文件，利用缓存 +COPY go.mod go.sum ./ +# 复制源代码 +COPY . . +#RUN go env -w GOPROXY=https://goproxy.cn,direct +RUN go env -w GOPROXY=http://172.20.32.233:30005/repository/hnxjy-goproxy/ +RUN go mod tidy && go mod download +# 构建应用 +RUN CGO_ENABLED=0 GOOS=linux go build -o remote-task-executor-cli + +# 第二阶段：运行阶段 +FROM 172.20.32.187/pipeline-service/pipeline-convert:running + +WORKDIR /app + +# 从构建阶段复制可执行文件 +COPY --from=builder /app/remote-task-executor-cli /app/ + +# 设置执行权限 +RUN chmod a+x /app/remote-task-executor-cli + +# 设置非 root 用户 +RUN addgroup -S appgroup && adduser -S appuser -G appgroup +# 健康检查 + +HEALTHCHECK --interval=30s --timeout=3s \ + CMD /app/remote-task-executor-cli healthcheck || exit 1 + +# 入口点 +ENTRYPOINT ["/app/remote-task-executor-cli", "run"] diff --git a/build-image.sh b/build-image.sh new file mode 100644 index 0000000..aa6613a --- /dev/null +++ b/build-image.sh @@ -0,0 +1,10 @@ +#!/bin/bash + +tag=$(date +'%Y%m%d%H%M') +image=172.20.32.187/pipeline-service/remote-task-cli:${tag} + +docker build -t ${image} . +docker push ${image} + +docker tag ${image} ccr.ccs.tencentyun.com/somunslotus/remote-task-cli:${tag} +docker push ccr.ccs.tencentyun.com/somunslotus/remote-task-cli:${tag} \ No newline at end of file diff --git a/cmd/root.go b/cmd/root.go index 8b8f23c..e7e95a3 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -2,6 +2,7 @@ package cmd import ( "github.com/spf13/cobra" + "os" ) var rootCmd = &cobra.Command{ @@ -12,6 +13,6 @@ var rootCmd = &cobra.Command{ func Execute() { if err := rootCmd.Execute(); err != nil { - panic(err) + os.Exit(2) } } diff --git a/cmd/run.go b/cmd/run.go index 8fb9e58..097ab9f 100644 --- a/cmd/run.go +++ b/cmd/run.go @@ -6,12 +6,14 @@ import ( "fmt" "github.com/spf13/cobra" "log" + "os" "remote-task-excutor-cli/pkg/config" "remote-task-excutor-cli/pkg/models" "remote-task-excutor-cli/pkg/runners" "remote-task-excutor-cli/pkg/service/auth" "remote-task-excutor-cli/pkg/service/preparation" "remote-task-excutor-cli/pkg/service/task" + "strings" "time" ) @@ -19,7 +21,7 @@ var runCmd = &cobra.Command{ Use: "run", Short: "Run an remote ML task", Long: `Execute an remote ML task with specified parameters`, - Run: func(cmd *cobra.Command, args []string) { + RunE: func(cmd *cobra.Command, args []string) error { // 创建参数结构体 fmt.Println("run called, args is", args) runConfig, err := parseParams() @@ -67,10 +69,11 @@ var runCmd = &cobra.Command{ if err := runner.RunTask(ctx, runConfig); err != nil { fmt.Println("Workflow failed with error:", err) - return + return err } fmt.Println("Workflow started successfully!") + return nil }, } @@ -79,7 +82,7 @@ func printParams(config *models.RunConfig) { fmt.Println("Starting remote ML task with configuration:") fmt.Printf("Code Config: %s\n", config.CodeConfig) fmt.Printf("Resource: %s\n", config.Resource) - fmt.Printf("Image: %s\n", config.Image) + fmt.Printf("Image: %d\n", config.Image) fmt.Printf("Command: %s\n", config.Command) fmt.Printf("Dataset: %s\n", config.Dataset) fmt.Printf("Model Name: %s\n", config.ModelName) @@ -104,12 +107,17 @@ func parseParams() (*models.RunConfig, error) { var r models.ResourceConfig var d models.DataResourceConfig var m models.DataResourceConfig - var rs models.RunConfig + var rs []string if err := json.Unmarshal([]byte(codeConfig), &c); err != nil { return nil, fmt.Errorf("failed to unmarshal codeConfig:%s, error:%v", codeConfig, err) } - + // 生成临时路径 + tempDir, err := os.MkdirTemp("", "code-*") + if err != nil { + return nil, fmt.Errorf("failed to create temp dir:%s, error:%v", tempDir, err) + } + c.MountPath = tempDir if err := json.Unmarshal([]byte(resource), &r); err != nil { return nil, fmt.Errorf("failed to unmarshal resourceConfig:%s, error:%v", resource, err) } @@ -118,14 +126,21 @@ func parseParams() (*models.RunConfig, error) { return nil, fmt.Errorf("failed to unmarshal datasetConfig:%s, error:%v", dataset, err) } - if err := json.Unmarshal([]byte(modelName), &m); err != nil { - return nil, fmt.Errorf("failed to unmarshal commandConfig:%s, error:%v", modelName, err) + if modelName != "" { + if err := json.Unmarshal([]byte(modelName), &m); err != nil { + return nil, fmt.Errorf("failed to unmarshal commandConfig:%s, error:%v", modelName, err) + } } if err := json.Unmarshal([]byte(runArgs), &rs); err != nil { return nil, fmt.Errorf("failed to unmarshal runArgsConfig:%s, error:%v", runArgs, err) } + params, err := ParseRunArgs(rs) + if err != nil { + return nil, fmt.Errorf("failed to parse runArgsConfig:%s, error:%v", runArgs, err) + } + return &models.RunConfig{ CodeConfig: c, Resource: r, @@ -133,11 +148,32 @@ func parseParams() (*models.RunConfig, error) { Command: command, Dataset: d, ModelName: m, - RunArgs: rs.RunArgs, + RunArgs: params, TaskOutput: taskOutput, }, nil } +// ParseRunArgs 解析运行参数字符串为键值对 +func ParseRunArgs(input []string) (map[string]string, error) { + result := make(map[string]string) + + if len(input) == 0 { + return result, nil + } + + for _, i := range input { + kv := strings.Split(i, "=") + if len(kv) != 2 { + return nil, fmt.Errorf("failed to parse runArgs: %s", i) + } + key := strings.TrimPrefix(strings.TrimSpace(kv[0]), "--") + value := strings.TrimSpace(kv[1]) + result[key] = value + } + + return result, nil +} + func init() { rootCmd.AddCommand(runCmd) @@ -150,8 +186,8 @@ func init() { runCmd.Flags().IntVarP(&image, "image", "i", 0, "运行镜像 (required)") - runCmd.Flags().IntVarP(&image, "resource_type", "i", 0, - "资源类型 (required)") + runCmd.Flags().StringVarP(&resourceType, "resource_type", "t", "", + "资源类型") runCmd.Flags().StringVarP(&command, "command", "m", "", "启动命令 (required)") diff --git a/go.mod b/go.mod index 86ca7cd..e4ffcb6 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,8 @@ module remote-task-excutor-cli -go 1.24.5 +go 1.24 + +toolchain go1.24.5 require ( github.com/go-git/go-git/v5 v5.16.2 @@ -8,7 +10,7 @@ require ( github.com/nacos-group/nacos-sdk-go/v2 v2.3.2 github.com/pkg/errors v0.9.1 github.com/spf13/cobra v1.9.1 - golang.org/x/sync v0.13.0 + golang.org/x/sync v0.15.0 ) require ( @@ -52,12 +54,13 @@ require ( github.com/golang/mock v1.6.0 // indirect github.com/golang/protobuf v1.5.4 // indirect github.com/golang/snappy v0.0.4 // indirect + github.com/google/go-containerregistry v0.20.6 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kevinburke/ssh_config v1.2.0 // indirect - github.com/klauspost/compress v1.17.11 // indirect + github.com/klauspost/compress v1.18.0 // indirect github.com/klauspost/pgzip v1.2.6 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect @@ -82,12 +85,12 @@ require ( go.uber.org/zap v1.21.0 // indirect golang.org/x/crypto v0.37.0 // indirect golang.org/x/net v0.39.0 // indirect - golang.org/x/sys v0.32.0 // indirect + golang.org/x/sys v0.33.0 // indirect golang.org/x/text v0.24.0 // indirect golang.org/x/time v0.1.0 // indirect google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect google.golang.org/grpc v1.56.3 // indirect - google.golang.org/protobuf v1.33.0 // indirect + google.golang.org/protobuf v1.36.3 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect gopkg.in/warnings.v0 v0.1.2 // indirect diff --git a/go.sum b/go.sum index e824216..bd8e713 100644 --- a/go.sum +++ b/go.sum @@ -227,6 +227,8 @@ github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-containerregistry v0.20.6 h1:cvWX87UxxLgaH76b4hIvya6Dzz9qHB31qAwjAohdSTU= +github.com/google/go-containerregistry v0.20.6/go.mod h1:T0x8MuoAoKX/873bkeSfLD2FAkwCDf9/HZgsFJ02E2Y= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs= github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0= @@ -270,6 +272,8 @@ github.com/klauspost/compress v1.4.1/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0 github.com/klauspost/compress v1.11.4/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs= github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc= github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/klauspost/cpuid v1.2.0/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= github.com/klauspost/pgzip v1.2.5/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs= github.com/klauspost/pgzip v1.2.6 h1:8RXeL5crjEUFnR2/Sn6GJNWtSQ3Dk8pq4CL3jvdDyjU= @@ -518,6 +522,8 @@ golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610= golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sync v0.15.0 h1:KWH3jNZsfyT6xfAfKiz6MRNmd46ByHDYaZ7KSkCtdW8= +golang.org/x/sync v0.15.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -572,6 +578,8 @@ golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20= golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw= +golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= @@ -734,6 +742,8 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI= google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= +google.golang.org/protobuf v1.36.3 h1:82DV7MYdb8anAVi3qge1wSnMDrnKK7ebr+I0hHRN1BU= +google.golang.org/protobuf v1.36.3/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/main.go b/main.go index 6892711..e1fab77 100644 --- a/main.go +++ b/main.go @@ -1,9 +1,21 @@ package main import ( + "fmt" + "os" "remote-task-excutor-cli/cmd" ) func main() { + os.Setenv("HTTP_PROXY", "http://172.20.32.233:3128") + os.Setenv("HTTPS_PROXY", "http://172.20.32.233:3128") + os.Setenv("NO_PROXY", "172.20.32.0/24") + fmt.Println("===== Environment Variables =====") + fmt.Printf("HTTP_PROXY: %s\n", os.Getenv("HTTP_PROXY")) + fmt.Printf("HTTPS_PROXY: %s\n", os.Getenv("HTTPS_PROXY")) + fmt.Printf("http_proxy: %s\n", os.Getenv("http_proxy")) + fmt.Printf("https_proxy: %s\n", os.Getenv("https_proxy")) + cmd.Execute() + //fmt.Println("Hello, world!") } diff --git a/pkg/client/http_client.go b/pkg/client/http_client.go index 36c5481..edd7214 100644 --- a/pkg/client/http_client.go +++ b/pkg/client/http_client.go @@ -11,6 +11,7 @@ import ( "net/url" "os" "path/filepath" + "strings" "time" ) @@ -97,7 +98,7 @@ func (c *HTTPClient) PostJSON(path string, data interface{}) ([]byte, error) { if err != nil { return nil, err } - + fmt.Println("request url, body is :", fullURL, string(jsonData)) req, err := http.NewRequest("POST", fullURL, bytes.NewBuffer(jsonData)) if err != nil { return nil, err @@ -129,6 +130,7 @@ func (c *HTTPClient) PostJSON(path string, data interface{}) ([]byte, error) { // files 参数现在支持每个字段名对应多个文件路径 func (c *HTTPClient) UploadFiles( path string, + filesPrefix string, files map[string][]string, // 字段名 -> 多个文件路径 formFields map[string]string, // 额外表单字段 ) ([]byte, error) { @@ -147,7 +149,12 @@ func (c *HTTPClient) UploadFiles( } defer file.Close() - part, err := writer.CreateFormFile(fieldName, filepath.Base(filePath)) + relativePath, err := removePathPrefix(filePath, filesPrefix) + if err != nil { + return nil, err + } + + part, err := writer.CreateFormFile(fieldName, url.PathEscape(relativePath)) if err != nil { return nil, fmt.Errorf("为文件 %s 创建表单字段失败: %w", filePath, err) } @@ -213,6 +220,31 @@ func (c *HTTPClient) UploadFiles( return io.ReadAll(resp.Body) } +func removePathPrefix(fullPath, prefix string) (string, error) { + // 使用系统文件分隔符标准化路径 + prefix = filepath.Clean(prefix) + string(filepath.Separator) + fullPath = filepath.Clean(fullPath) + + // 检查路径是否有指定的前缀 + if strings.HasPrefix(fullPath, prefix) { + // 返回去除前缀的路径 + return fullPath[len(prefix):], nil + } + // 如果不匹配，返回原始路径（或者可以根据需要返回错误） + return "", fmt.Errorf("错误的filepath和prefix, filepath:%v, prefix:%v", fullPath, prefix) +} + +func (c *HTTPClient) encodeFilepath(paths []string) []string { + var result []string + for _, path := range paths { + // 编码文件路径,使用urlencoded编码 + encodedPath := url.PathEscape(path) + result = append(result, encodedPath) + } + fmt.Println("encodeFilepath result: ", result) + return result +} + // DownloadFile 下载文件到本地路径（支持大文件流式处理） // localPath：保存文件的本地路径（包含文件名） // queryParams：可选查询参数 @@ -262,9 +294,12 @@ func (c *HTTPClient) DownloadFile(path, localPath string, queryParams ...map[str } } + os.MkdirAll(localPath, 0755) filename := extractFilenameFromDisposition(resp.Header.Get("Content-Disposition")) if filename == "" { filename = filepath.Join(localPath, "file.zip") + } else { + filename = filepath.Join(localPath, filename) } // 创建本地文件 outFile, err := os.Create(filename) diff --git a/pkg/config/config.go b/pkg/config/config.go index d4b9f10..bbcff88 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -39,6 +39,7 @@ type APIConfig struct { Username string `json:"username" yaml:"username"` Password string `json:"password" yaml:"password"` BaseURL string `json:"baseUrl" yaml:"baseUrl"` + TestURL string `json:"testUrl" yaml:"testUrl"` TokenTTL int `json:"tokenTtl" yaml:"tokenTtl"` // token 有效期（秒） Timeout int `json:"timeout" yaml:"timeout"` } @@ -54,6 +55,16 @@ type ConfigReader struct { group string } +var DefaultApiConfig = &APIConfig{ + AuthURL: "jcc-admin/admin/login", + Username: "hnxjy-super", + Password: "h1n2x3j4y5@", + BaseURL: "https://jcc.jointcloud.net/", + TestURL: "", + TokenTTL: 0, + Timeout: 10, +} + // NewConfigReader 创建新的 Nacos 配置读取器 func NewConfigReader(ctx context.Context, cfg NacosConfig) (*ConfigReader, error) { // 创建 ServerConfig @@ -67,7 +78,7 @@ func NewConfigReader(ctx context.Context, cfg NacosConfig) (*ConfigReader, error // 创建 ClientConfig clientConfig := constant.ClientConfig{ - NamespaceId: cfg.NamespaceID, + NamespaceId: "", TimeoutMs: 5000, NotLoadCacheAtStart: true, LogDir: cfg.LogDir, @@ -109,34 +120,34 @@ func NewConfigReader(ctx context.Context, cfg NacosConfig) (*ConfigReader, error // loadConfig 从Nacos加载配置（带上下文） func (r *ConfigReader) loadConfig(ctx context.Context) error { // 使用上下文限制超时 - content, err := r.client.GetConfig(vo.ConfigParam{ - DataId: r.dataID, - Group: r.group, - }) - - if err != nil { - return errors.Wrap(err, "获取Nacos配置失败") - } - - if content == "" { - return errors.New("从Nacos获取的配置为空") - } - - var apiCfg APIConfig - if err := json.Unmarshal([]byte(content), &apiCfg); err != nil { - return errors.Wrapf(err, "解析配置失败: %s", content) - } - - // 设置默认值 - if apiCfg.TokenTTL == 0 { - apiCfg.TokenTTL = 3600 // 默认1小时 - } - - // 更新配置 - r.cfgMutex.Lock() - r.apiCfg = &apiCfg - r.cfgMutex.Unlock() - + //content, err := r.client.GetConfig(vo.ConfigParam{ + // DataId: r.dataID, + // Group: r.group, + //}) + // + //if err != nil { + // return errors.Wrap(err, "获取Nacos配置失败") + //} + // + //if content == "" { + // return errors.New("从Nacos获取的配置为空") + //} + // + //var apiCfg APIConfig + //if err := yaml.Unmarshal([]byte(content), &apiCfg); err != nil { + // return errors.Wrapf(err, "解析配置失败: %s", content) + //} + // + //// 设置默认值 + //if apiCfg.TokenTTL == 0 { + // apiCfg.TokenTTL = 3600 // 默认1小时 + //} + // + //// 更新配置 + //r.cfgMutex.Lock() + //r.apiCfg = &apiCfg + //r.cfgMutex.Unlock() + r.apiCfg = DefaultApiConfig return nil } diff --git a/pkg/models/auth.go b/pkg/models/auth.go index e479cfe..b152d78 100644 --- a/pkg/models/auth.go +++ b/pkg/models/auth.go @@ -10,15 +10,15 @@ type AuthResponse struct { // AuthData 认证数据主体 type AuthData struct { TokenHead string `json:"tokenHead"` - ExpiresIn int `json:"expiresIn"` // 注意：改为整型 + ExpiresIn string `json:"expiresIn"` JsmUserInfo UserInfo `json:"jsmUserInfo"` - TokenTimeout int `json:"tokenTimeout"` // 注意：改为整型 + TokenTimeout string `json:"tokenTimeout"` Token string `json:"token"` } // UserInfo 用户信息 type UserInfo struct { - Code int `json:"code"` // 注意：改为整型 + Code string `json:"code"` // 注意：改为整型 Message string `json:"message"` Data UserData `json:"data"` } diff --git a/pkg/models/params.go b/pkg/models/params.go index 0987eed..d99cc3c 100644 --- a/pkg/models/params.go +++ b/pkg/models/params.go @@ -1,40 +1,13 @@ package models -import ( - "fmt" - "strings" -) - // RunConfig 包含所有运行参数 type RunConfig struct { CodeConfig CodeConfig `json:"code_config"` Resource ResourceConfig `json:"resource"` Image int `json:"image"` Command string `json:"command"` - RunArgs []string `json:"run_args"` + RunArgs map[string]string `json:"run_args"` Dataset DataResourceConfig `json:"dataset"` ModelName DataResourceConfig `json:"model_name"` TaskOutput string `json:"task_output"` } - -// ParseRunArgs 解析运行参数字符串为键值对 -func ParseRunArgs(input string) (map[string]string, error) { - result := make(map[string]string) - - if input == "" { - return result, nil - } - - pairs := strings.Split(input, ",") - for _, pair := range pairs { - kv := strings.SplitN(pair, "=", 2) - if len(kv) != 2 { - return nil, fmt.Errorf("invalid key-value pair: %s", pair) - } - key := strings.TrimSpace(kv[0]) - value := strings.TrimSpace(kv[1]) - result[key] = value - } - - return result, nil -} diff --git a/pkg/models/preparation.go b/pkg/models/preparation.go index 3aaedce..d4054b9 100644 --- a/pkg/models/preparation.go +++ b/pkg/models/preparation.go @@ -37,11 +37,11 @@ type BucketID int type StorageID int type PackageCreateLoadInfo struct { - UserID int `json:"userID" binding:"required"` - BucketID int `json:"bucketID" binding:"required"` - Name string `json:"name" binding:"required"` - LoadTo int `json:"loadTo"` - LoadToPath []string `json:"loadToPath"` + UserID int `json:"userID" binding:"required"` + BucketID int `json:"bucketID" binding:"required"` + Name string `json:"name" binding:"required"` + //LoadTo int `json:"loadTo"` + //LoadToPath []string `json:"loadToPath"` } type PackageCreateResponse struct { @@ -208,3 +208,8 @@ type BindResultSet struct { BindDatasetID int `json:"bind_dataset_id"` BindModelID int `json:"bind_model_id"` } + +type UploadFileConfig struct { + FilePrefix string + Files []string +} diff --git a/pkg/models/task.go b/pkg/models/task.go index e77ae7a..c845c6e 100644 --- a/pkg/models/task.go +++ b/pkg/models/task.go @@ -27,7 +27,7 @@ type Job struct { // JobFiles 表示任务关联的文件 type JobFiles struct { Dataset FileBinding `json:"dataset"` - Model FileBinding `json:"model"` + Model FileBinding `json:"model,omitempty"` Image ImageBinding `json:"image"` } @@ -118,8 +118,8 @@ type TaskDetailResponse struct { type TaskDetailData struct { Name string `json:"name"` // 任务名称 Description string `json:"description"` // 任务描述 - StartTime time.Time `json:"startTime"` // 开始时间 - EndTime time.Time `json:"endTime"` // 结束时间 + StartTime string `json:"startTime"` // 开始时间 + EndTime string `json:"endTime"` // 结束时间 Strategy int `json:"strategy"` // 策略 SynergyStatus int `json:"synergyStatus"` // 协同状态 ClusterInfos []ClusterItem `json:"clusterInfos"` // 集群信息列表 diff --git a/pkg/runners/task_runner.go b/pkg/runners/task_runner.go index 2846938..8acb339 100644 --- a/pkg/runners/task_runner.go +++ b/pkg/runners/task_runner.go @@ -28,12 +28,12 @@ func (r *taskRunner) RunTask(ctx context.Context, config *models.RunConfig) erro if err != nil { return err } - + fmt.Println("获取token成功") clusterID, err := r.authService.GetClusterID(ctx, "openI") if err != nil { return err } - + fmt.Println("获取clusterID成功：", clusterID) // 数据准备 bindResultSet, err := r.preparationService.PrepareAll(ctx, authData, config, clusterID) if err != nil { @@ -44,13 +44,18 @@ func (r *taskRunner) RunTask(ctx context.Context, config *models.RunConfig) erro if err != nil { return err } - + fmt.Println("任务提交成功，开始查询任务状态") + time.Sleep(5 * time.Second) + go r.RecordLogFile(ctx, authData, jobSetID) // 轮询任务状态 - err = r.PollTaskStatusWithBackoff(ctx, authData, jobSetID) + resp, err := r.PollTaskStatusWithBackoff(ctx, authData, jobSetID) if err != nil { return err } + if resp.Data.SubTaskInfos[0].Status != "Completed" { + return fmt.Errorf("远程任务执行失败:") + } // 获取任务结果 err = r.taskRunService.GetTaskResult(ctx, authData, config.TaskOutput, jobSetID) if err != nil { @@ -60,11 +65,21 @@ func (r *taskRunner) RunTask(ctx context.Context, config *models.RunConfig) erro return nil } -func (r *taskRunner) PollTaskStatusWithBackoff(ctx context.Context, authData *models.AuthData, jobSetID string) error { +func (r *taskRunner) RecordLogFile(ctx context.Context, authData *models.AuthData, jobSetID string) error { + for { + content, _ := r.taskRunService.GetTaskLogs(ctx, authData, jobSetID) + if len(content) == 0 { + time.Sleep(5 * time.Second) + } + fmt.Println("get log content is ", content) + } +} + +func (r *taskRunner) PollTaskStatusWithBackoff(ctx context.Context, authData *models.AuthData, jobSetID string) (*models.TaskDetailResponse, error) { const ( baseInterval = 5 * time.Second maxInterval = 1 * time.Minute - maxAttempts = 10 // 最多尝试10次 + maxAttempts = 5 // 最多尝试10次 ) var ( @@ -73,26 +88,26 @@ func (r *taskRunner) PollTaskStatusWithBackoff(ctx context.Context, authData *mo ) for { - attemptCount++ if attemptCount > maxAttempts { - return fmt.Errorf("超出最大查询次数(%d)", maxAttempts) + return nil, fmt.Errorf("超出最大查询次数(%d)", maxAttempts) } // 检查上下文是否被取消 if ctx.Err() != nil { - return ctx.Err() + return nil, ctx.Err() } // 获取任务状态 status, err := r.taskRunService.GetTaskStatus(ctx, authData, jobSetID) if err != nil { + attemptCount++ // 指数回退 backoff = min(backoff*2, maxInterval) sleepTime := backoff + time.Duration(rand.Int63n(int64(backoff/2))) // 随机抖动避免同步 fmt.Printf("任务查询失败(%s), 将在 %s 后重试\n", err, sleepTime) select { case <-ctx.Done(): - return ctx.Err() + return nil, ctx.Err() case <-time.After(sleepTime): continue } @@ -103,17 +118,17 @@ func (r *taskRunner) PollTaskStatusWithBackoff(ctx context.Context, authData *mo // 检查任务状态 switch { - case IsCompleted(status): - fmt.Println("任务成功完成!") - return nil + case IsCompleted(status.Data.SubTaskInfos[0].Status): + fmt.Println("任务成功完成!, 完成状态：", status) + return status, nil default: - fmt.Println("任务运行中: %s", status) + fmt.Println("任务运行中: ", status) } // 等待下一次轮询 select { case <-ctx.Done(): - return ctx.Err() + return nil, ctx.Err() case <-time.After(baseInterval): } } diff --git a/pkg/service/auth/auth_service.go b/pkg/service/auth/auth_service.go index 287ded6..03980e2 100644 --- a/pkg/service/auth/auth_service.go +++ b/pkg/service/auth/auth_service.go @@ -29,7 +29,8 @@ func NewTokenService(cfg *config.APIConfig) service.AuthService { authURL: cfg.AuthURL, username: cfg.Username, password: cfg.Password, - httpClient: client.NewHTTPClient(cfg.AuthURL, time.Second*10), + httpClient: client.NewHTTPClient(cfg.BaseURL, time.Duration(cfg.Timeout)*time.Second), + authData: &models.AuthData{}, } } @@ -81,7 +82,11 @@ func (ts *TokenService) GetClusterID(ctx context.Context, label string) (string, // 查询集群ID ts.httpClient.Headers["Authorization"] = "Bearer " + token.Token - resp, err := ts.httpClient.Get("/pcm/v1/adapter/cluster/getClusterBaseInfo", nil) + params := map[string]string{ + "pageNum": "1", + "pageSize": "30", + } + resp, err := ts.httpClient.Get("/pcm/v1/adapter/cluster/getClusterBaseInfo", params) if err != nil { return "", err } diff --git a/pkg/service/interface.go b/pkg/service/interface.go index 7925066..f51f1c2 100644 --- a/pkg/service/interface.go +++ b/pkg/service/interface.go @@ -22,7 +22,7 @@ type PreparationService interface { type TaskService interface { SubmitTask(ctx context.Context, authData *models.AuthData, config *models.RunConfig, clusterID string, bindResultSet *models.BindResultSet) (string, error) - GetTaskStatus(ctx context.Context, authData *models.AuthData, jobSetID string) (string, error) + GetTaskStatus(ctx context.Context, authData *models.AuthData, jobSetID string) (*models.TaskDetailResponse, error) GetTaskResult(ctx context.Context, authData *models.AuthData, path string, jobSetID string) error GetTaskLogs(ctx context.Context, authData *models.AuthData, jobSetID string) (string, error) } diff --git a/pkg/service/preparation/preparation_service.go b/pkg/service/preparation/preparation_service.go index f0e0a50..36b4ec7 100644 --- a/pkg/service/preparation/preparation_service.go +++ b/pkg/service/preparation/preparation_service.go @@ -28,10 +28,12 @@ const ( ImageType = "image" CodeType = "code" BindingType = "binding" + TaskBindingType = "Binding" ImageCategory = "image" RootPath = "/" ImageClassifyType = "image_classification" PytorchType = "pytorch" + CreateResourceUri = "/apis/jsm//app/submit" ) type preparationService struct { @@ -57,7 +59,10 @@ func (s *preparationService) PrepareAll(ctx context.Context, authData *models.Au codeChan := make(chan result, 1) datasetChan := make(chan result, 1) - modelChan := make(chan result, 1) + var modelChan chan result + if config.ModelName.Path != "" { + modelChan = make(chan result, 1) + } // 并发执行代码准备 errGroup.Go(func() error { @@ -73,19 +78,26 @@ func (s *preparationService) PrepareAll(ctx context.Context, authData *models.Au return err }) - // 并发执行模型准备 - errGroup.Go(func() error { - id, err := s.PrepareModel(ctx, authData, config.ModelName, clusterID) - modelChan <- result{id, err} - return err - }) + if config.ModelName.Path != "" { + errGroup.Go(func() error { + id, err := s.PrepareModel(ctx, authData, config.ModelName, clusterID) + modelChan <- result{id, err} + return err + }) + } // 等待所有任务完成 if err := errGroup.Wait(); err != nil { // 有错误发生，但我们需要收集所有结果 codeRes := <-codeChan datasetRes := <-datasetChan - modelRes := <-modelChan + var modelRes result + + if modelChan != nil { + modelRes = <-modelChan + } else { + modelRes = result{id: 0, err: nil} // 默认值 + } return &models.BindResultSet{ BindCodeID: codeRes.id, @@ -97,7 +109,12 @@ func (s *preparationService) PrepareAll(ctx context.Context, authData *models.Au // 所有任务成功完成 codeRes := <-codeChan datasetRes := <-datasetChan - modelRes := <-modelChan + var modelRes result + if modelChan != nil { + modelRes = <-modelChan + } else { + modelRes = result{id: 0, err: nil} + } return &models.BindResultSet{ BindCodeID: codeRes.id, @@ -124,31 +141,52 @@ func aggregateErrors(errs ...error) error { func (s *preparationService) PrepareCode(ctx context.Context, authData *models.AuthData, config *models.RunConfig, clusterID string) (int, error) { + fmt.Println("开始准备代码") resul, err := s.uploadCode(ctx, authData, config.CodeConfig) if err != nil { return 0, err } - return s.bindCode(ctx, authData, config, resul, clusterID) + fmt.Println("代码上传成功，开始代码定版") + id, err := s.bindCode(ctx, authData, config, resul, clusterID) + if err != nil { + return 0, err + } + fmt.Println("代码定版成功") + return id, nil } func (s *preparationService) PrepareDataset(ctx context.Context, authData *models.AuthData, config models.DataResourceConfig, clusterID string) (int, error) { + fmt.Println("开始准备数据集") // 1 上传数据集到数据集仓库 result, err := s.uploadFiles(ctx, authData, config, DatasetType) if err != nil { return 0, fmt.Errorf("上传数据集失败: %w", err) } + fmt.Println("数据集上传成功，开始数据集定版") //2 绑定数据集到用户的bucket - return s.bindDataset(ctx, authData, config, result, clusterID) + id, err := s.bindDataset(ctx, authData, config, result, clusterID) + if err != nil { + return 0, err + } + fmt.Println("数据集定版成功") + return id, nil } func (s *preparationService) PrepareModel(ctx context.Context, authData *models.AuthData, config models.DataResourceConfig, clusterID string) (int, error) { + fmt.Println("开始准备模型") result, err := s.uploadFiles(ctx, authData, config, ModelType) if err != nil { return 0, fmt.Errorf("上传模型失败: %w", err) } - return s.bindModel(ctx, authData, config, result, clusterID) + fmt.Println("模型上传成功，开始模型定版") + id, err := s.bindModel(ctx, authData, config, result, clusterID) + if err != nil { + return 0, err + } + fmt.Println("模型定版成功") + return id, nil } func (s *preparationService) bindCode(ctx context.Context, authData *models.AuthData, config *models.RunConfig, @@ -187,7 +225,8 @@ func (s *preparationService) bindCode(ctx context.Context, authData *models.Auth }, } s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token - resp, err := s.httpClient.PostJSON("/app/submit", request) + resp, err := s.httpClient.PostJSON(CreateResourceUri, request) + fmt.Println("代码绑定结果：", string(resp)) if err != nil { return 0, fmt.Errorf("绑定代码失败: %w", err) } @@ -196,7 +235,7 @@ func (s *preparationService) bindCode(ctx context.Context, authData *models.Auth func (s *preparationService) getStartCommandObjectID(config *models.RunConfig, uploadResult *models.PackageCreateResponse) (int, error) { for _, file := range uploadResult.Data.Objects { - if (RootPath + strings.Trim(config.Command, " ")) == file.Path { + if (strings.Trim(config.Command, " ")) == file.Path { return file.ObjectID, nil } } @@ -235,7 +274,8 @@ func (s *preparationService) bindModel(ctx context.Context, authData *models.Aut }, } s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token - resp, err := s.httpClient.PostJSON("/app/submit", request) + resp, err := s.httpClient.PostJSON(CreateResourceUri, request) + fmt.Println("模型绑定结果：", string(resp)) if err != nil { return 0, fmt.Errorf("绑定模型失败: %w", err) } @@ -244,7 +284,7 @@ func (s *preparationService) bindModel(ctx context.Context, authData *models.Aut func (s *preparationService) uploadFiles(ctx context.Context, authData *models.AuthData, config models.DataResourceConfig, uploadType string) (*models.PackageCreateResponse, error) { // 1 获取数据集路径下的所有文件 遍历config.MountPath下的所有文件，获取路径和文件名，组成文件map[string][]string - files, err := GetAllFiles(config.MountPath) + filesConfig, err := GetAllFiles(config.MountPath) if err != nil { return nil, fmt.Errorf("获取路径下的所有文件失败: %w", err) } @@ -259,19 +299,20 @@ func (s *preparationService) uploadFiles(ctx context.Context, authData *models.A Name: config.Name + "-" + GenerateUniqueID(), } - return s.doUpload(ctx, authData, files, info) + return s.doUpload(ctx, authData, filesConfig, info) } -func (s *preparationService) doUpload(ctx context.Context, authData *models.AuthData, files []string, info models.PackageCreateLoadInfo) (*models.PackageCreateResponse, error) { +func (s *preparationService) doUpload(ctx context.Context, authData *models.AuthData, filesConfig *models.UploadFileConfig, info models.PackageCreateLoadInfo) (*models.PackageCreateResponse, error) { infoJson, err := json.Marshal(info) if err != nil { return nil, fmt.Errorf("序列化PackageCreateLoadInfo失败: %w", err) } s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token - resp, err := s.httpClient.UploadFiles("package/createLoad", + resp, err := s.httpClient.UploadFiles("/apis/jcs/package/createLoad", + filesConfig.FilePrefix, map[string][]string{ - "files": s.encodeFilepath(files), + "files": filesConfig.Files, //s.encodeFilepath(files), }, map[string]string{ "info": string(infoJson), @@ -285,6 +326,9 @@ func (s *preparationService) doUpload(ctx context.Context, authData *models.Auth if err != nil { return nil, fmt.Errorf("解析上传结果失败: %w", err) } + if result.Code != "OK" { + return nil, fmt.Errorf("上传失败：%v-%v", result.Code, result.Message) + } return &result, nil } @@ -300,7 +344,7 @@ func (s *preparationService) uploadCode(ctx context.Context, authData *models.Au info := models.PackageCreateLoadInfo{ UserID: authData.JsmUserInfo.Data.UserID, BucketID: authData.JsmUserInfo.Data.Buckets.Code, - Name: config.CodeRepoName + "-" + GenerateUniqueID(), + Name: "code" + "-" + GenerateUniqueID(), } return s.doUpload(ctx, authData, files, info) } @@ -354,7 +398,8 @@ func (s *preparationService) bindDataset(ctx context.Context, authData *models.A }, } s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token - resp, err := s.httpClient.PostJSON("/app/submit", request) + resp, err := s.httpClient.PostJSON(CreateResourceUri, request) + fmt.Println("数据集绑定结果：", string(resp)) if err != nil { return 0, fmt.Errorf("绑定数据集失败: %w", err) } @@ -373,7 +418,7 @@ func (s *preparationService) getBindingID(response []byte) (int, error) { } if result.BindingID <= 0 { - return 0, fmt.Errorf("绑定数据集失败: %s", tmp.Message) + return 0, fmt.Errorf("绑定资源失败: %s", tmp.Message) } return result.BindingID, nil @@ -412,7 +457,7 @@ func getObjectIDS(uploadResult *models.PackageCreateResponse) []int { } // GetAllFiles 获取指定目录下的所有文件绝对路径 -func GetAllFiles(rootDir string) ([]string, error) { +func GetAllFiles(rootDir string) (*models.UploadFileConfig, error) { // 获取绝对路径 absRoot, err := filepath.Abs(rootDir) if err != nil { @@ -426,6 +471,11 @@ func GetAllFiles(rootDir string) ([]string, error) { return fmt.Errorf("访问路径 %s 失败: %w", path, err) } + // 排除 .git 目录及其所有内容 + if info.IsDir() && info.Name() == ".git" { + return filepath.SkipDir // 跳过整个目录 + } + if !info.IsDir() { // 确保路径是绝对路径 absPath, err := filepath.Abs(path) @@ -441,7 +491,10 @@ func GetAllFiles(rootDir string) ([]string, error) { return nil, fmt.Errorf("遍历目录失败: %w", err) } - return files, nil + return &models.UploadFileConfig{ + FilePrefix: rootDir, + Files: files, + }, nil } func (s *preparationService) encodeFilepath(paths []string) []string { diff --git a/pkg/service/task/task_service.go b/pkg/service/task/task_service.go index 206fac4..595ace2 100644 --- a/pkg/service/task/task_service.go +++ b/pkg/service/task/task_service.go @@ -38,12 +38,13 @@ func (s *taskService) SubmitTask(ctx context.Context, authData *models.AuthData, clusterID string, bindResultSet *models.BindResultSet) (string, error) { submitTaskReq := s.buildSubmitTaskReq(ctx, authData, config, clusterID, bindResultSet) s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token + fmt.Printf("任务提交参数：%+v", submitTaskReq) resp, err := s.httpClient.PostJSON("/jsm/jobSet/submit", submitTaskReq) if err != nil { fmt.Println("Submit task failed: ", err) return "", err } - + fmt.Println("提交任务结果：", string(resp)) var submitTaskResp models.SubmitTaskResponse if err := json.Unmarshal(resp, &submitTaskResp); err != nil { fmt.Println("Submit task response unmarshal failed: ", err) @@ -53,34 +54,41 @@ func (s *taskService) SubmitTask(ctx context.Context, authData *models.AuthData, fmt.Println("Submit task failed: ", submitTaskResp.Code) return "", fmt.Errorf("submit task failed: %s", submitTaskResp.Code) } - fmt.Println("Submit task result: ", resp) + fmt.Println("Submit task result: ", string(resp)) return submitTaskResp.Data.JobSetID, nil } -func (s *taskService) GetTaskStatus(ctx context.Context, authData *models.AuthData, jobSetID string) (string, error) { +func (s *taskService) GetTaskStatus(ctx context.Context, authData *models.AuthData, jobSetID string) (*models.TaskDetailResponse, error) { s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token params := map[string]string{ JobSetIDKey: jobSetID, LocalJobIDKey: MainTaskID, } - resp, err := s.httpClient.Get("/jsm/jobMgr/detail", params) - if err != nil { - fmt.Println("Get task status failed: ", err) - return "", err - } - var taskStatus models.TaskDetailResponse - if err := json.Unmarshal(resp, &taskStatus); err != nil { - fmt.Println("Get task status response unmarshal failed: ", err) - return "", err - } + for { + resp, err := s.httpClient.Get("/jsm/jobMgr/detail", params) + if err != nil { + fmt.Println("Get task status failed: ", err) + return nil, err + } - if taskStatus.Code != ResponseOK { - fmt.Println("Get task status failed: ", taskStatus.Code) - return "", fmt.Errorf("get task status failed: code: %s, message: %s", taskStatus.Code, taskStatus.Message) + var taskStatus models.TaskDetailResponse + if err := json.Unmarshal(resp, &taskStatus); err != nil { + fmt.Println("Get task status response unmarshal failed: ", err) + return nil, err + } + + if taskStatus.Code != ResponseOK { + fmt.Println("Get task status failed: ", taskStatus.Code) + return nil, fmt.Errorf("get task status failed: code: %s, message: %s", taskStatus.Code, taskStatus.Message) + } + fmt.Println("Get task status result: ", string(resp)) + if len(taskStatus.Data.SubTaskInfos) == 0 { + time.Sleep(5 * time.Second) + continue + } + return &taskStatus, nil } - fmt.Println("Get task status result: ", resp) - return taskStatus.Data.SubTaskInfos[0].Status, nil } func (s *taskService) GetTaskLogs(ctx context.Context, authData *models.AuthData, jobSetID string) (string, error) { @@ -98,6 +106,7 @@ func (s *taskService) GetTaskLogs(ctx context.Context, authData *models.AuthData var taskLogResponse models.TaskLogResponse if err := json.Unmarshal(resp, &taskLogResponse); err != nil { fmt.Println("Get task logs response unmarshal failed: ", err) + return "", err } @@ -125,6 +134,7 @@ func (s *taskService) GetTaskResult(ctx context.Context, authData *models.AuthDa fmt.Println("Unarchive task result failed: ", err) return err } + // 删除filename文件 if err := os.Remove(filename); err != nil { fmt.Println("Remove task result failed: ", err) @@ -137,8 +147,8 @@ func (s *taskService) GetTaskResult(ctx context.Context, authData *models.AuthDa func (s *taskService) downloadTaskResult(ctx context.Context, authData *models.AuthData, path string, packageID int) (string, error) { s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token params := map[string]string{ - "packageId": fmt.Sprintf("%s", packageID), - "userId": fmt.Sprintf("%s", authData.JsmUserInfo.Data.UserID), + "packageID": fmt.Sprintf("%d", packageID), + "userID": fmt.Sprintf("%d", authData.JsmUserInfo.Data.UserID), } filename, err := s.httpClient.DownloadFile("/jcs/v1/package/download", path, params) if err != nil { @@ -153,31 +163,52 @@ func (s *taskService) getTaskResultDetail(ctx context.Context, authData *models. s.httpClient.Headers["Authorization"] = "Bearer " + authData.Token queryTaskResultReq := models.TaskQueryRequest{ JobSetID: jobSetID, - LocalJobID: DataReturnTaskID, + LocalJobID: MainTaskID, } - resp, err := s.httpClient.PostJSON("/jsm/jobMgr/result", queryTaskResultReq) - if err != nil { - fmt.Println("Get task result details failed: ", err) - return nil, err + maxAttemps := 3 + attemps := 0 + for { + resp, err := s.httpClient.PostJSON("/jsm/jobMgr/result", queryTaskResultReq) + fmt.Println("get result details resp:", string(resp)) + if err != nil { + fmt.Println("Get task result details failed: ", err) + return nil, err + } + + var taskResultDetails models.TaskResultDetailResponse + if err := json.Unmarshal(resp, &taskResultDetails); err != nil { + fmt.Println("Get task result details response unmarshal failed: ", err) + return nil, err + } + + if taskResultDetails.Code != ResponseOK { + if taskResultDetails.Message == "query pcm job: record not found" { + time.Sleep(5 * time.Second) + continue + } else { + if attemps > maxAttemps { + fmt.Println("Get task result details failed: ", taskResultDetails.Code) + return nil, fmt.Errorf("get task result details failed: %s", taskResultDetails.Code) + } + attemps++ + time.Sleep(5 * time.Second) + } + } + + if taskResultDetails.Data.PcmJobData.Status == "failed" { // && taskResultDetails.Data.PcmJobData.ErrorMsg == "data return job id is empty" { + time.Sleep(5 * time.Second) + continue + } + + return &taskResultDetails.Data, nil } - var taskResultDetails models.TaskResultDetailResponse - if err := json.Unmarshal(resp, &taskResultDetails); err != nil { - fmt.Println("Get task result details response unmarshal failed: ", err) - return nil, err - } - - if taskResultDetails.Code != ResponseOK { - fmt.Println("Get task result details failed: ", taskResultDetails.Code) - return nil, fmt.Errorf("get task result details failed: %s", taskResultDetails.Code) - } - - return &taskResultDetails.Data, nil } func (s *taskService) buildSubmitTaskReq(ctx context.Context, authData *models.AuthData, config *models.RunConfig, clusterID string, bindResultSet *models.BindResultSet) models.SubtaskRequest { + return models.SubtaskRequest{ UserID: authData.JsmUserInfo.Data.UserID, JobSetInfo: models.JobSetInfo{ @@ -189,15 +220,15 @@ func (s *taskService) buildSubmitTaskReq(ctx context.Context, authData *models.A Type: AITaskType, Files: &models.JobFiles{ Dataset: models.FileBinding{ - Type: preparation.DatasetType, + Type: preparation.TaskBindingType, BindingID: bindResultSet.BindDatasetID, }, Model: models.FileBinding{ - Type: preparation.ModelType, + Type: preparation.TaskBindingType, BindingID: bindResultSet.BindModelID, }, Image: models.ImageBinding{ - Type: preparation.ImageType, + Type: "Image", ImageID: config.Image, }, }, @@ -207,11 +238,11 @@ func (s *taskService) buildSubmitTaskReq(ctx context.Context, authData *models.A { ClusterID: clusterID, Runtime: models.Runtime{ - Envs: nil, - Params: nil, + Envs: config.RunArgs, + Params: config.RunArgs, }, Code: models.CodeInfo{ - Type: preparation.CodeType, + Type: preparation.TaskBindingType, BindingID: bindResultSet.BindCodeID, }, Resources: config.Resource.Resources, @@ -226,7 +257,7 @@ func (s *taskService) buildSubmitTaskReq(ctx context.Context, authData *models.A { TargetJobID: MainTaskID, InputParams: models.InputParams{ - PackageName: "name", + PackageName: "Name", ClusterID: "ClusterID", Output: "Output", }, diff --git a/remote-task-excutor-cli.exe b/remote-task-excutor-cli.exe deleted file mode 100644 index d6095c7..0000000 Binary files a/remote-task-excutor-cli.exe and /dev/null differ