From 113d7b7da8282b2eccc0aeb714fdd7771c339e9b Mon Sep 17 00:00:00 2001 From: somunslotus Date: Tue, 25 Feb 2025 15:23:27 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B6=85=E5=8F=82=E5=AF=BB=E4=BC=98workflow?= =?UTF-8?q?=E8=BD=AC=E6=8D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- build_for_181.sh | 4 +- docs/env/loki-log/loki/loki/values.yaml | 6 +- internal/handler/convert_handler_new.go | 327 +++++++++++++++++++++++- internal/router/api_router.go | 1 + internal/service/parse/parse.go | 2 + pipeline-convert.yaml | 14 +- 6 files changed, 344 insertions(+), 10 deletions(-) diff --git a/build_for_181.sh b/build_for_181.sh index 0435b92..69d8922 100644 --- a/build_for_181.sh +++ b/build_for_181.sh @@ -63,8 +63,8 @@ fi echo "################### deploy start ###################" -scp pipeline-convert.yaml root@172.20.32.181:/home/deploy/script/pipeline-convert/pipeline-convert.yaml -ssh root@172.20.32.181 "unset http_proxy && unset https_proxy && kubectl apply -f /home/deploy/script/pipeline-convert/pipeline-convert.yaml" +scp pipeline-convert.yaml root@172.20.32.197:/home/deploy/script/pipeline-convert/pipeline-convert.yaml +ssh root@172.20.32.197 "unset http_proxy && unset https_proxy && kubectl apply -f /home/deploy/script/pipeline-convert/pipeline-convert.yaml" sed -i "s#${image}#pipeline-image#g" pipeline-convert.yaml if [ "$?" -ne "0" ];then echo "deploy fail" diff --git a/docs/env/loki-log/loki/loki/values.yaml b/docs/env/loki-log/loki/loki/values.yaml index c3d6249..85f2e0d 100644 --- a/docs/env/loki-log/loki/loki/values.yaml +++ b/docs/env/loki-log/loki/loki/values.yaml @@ -39,7 +39,7 @@ loki: timeoutSeconds: 1 image: # -- The Docker registry - registry: docker.io + registry: 172.20.32.187/pipeline-service # -- Docker image repository repository: grafana/loki # -- Overrides the image tag whose default is the chart's appVersion @@ -726,7 +726,7 @@ monitoring: # -- Image to use for loki canary image: # -- The Docker registry - registry: docker.io + registry: 172.20.32.187/pipeline-service # -- Docker image repository repository: grafana/loki-canary # -- Overrides the image tag whose default is the chart's appVersion @@ -1315,7 +1315,7 @@ gateway: type: RollingUpdate image: # -- The Docker registry for the gateway image - registry: docker.io + registry: 172.20.32.187/pipeline-service # -- The gateway image repository repository: nginxinc/nginx-unprivileged # -- The gateway image tag diff --git a/internal/handler/convert_handler_new.go b/internal/handler/convert_handler_new.go index 4ef7cc4..555b217 100644 --- a/internal/handler/convert_handler_new.go +++ b/internal/handler/convert_handler_new.go @@ -11,6 +11,7 @@ import ( "pipeline-convert/internal/service" "pipeline-convert/internal/service/parse" "pipeline-convert/internal/utils/stringutils" + "strconv" "strings" ) @@ -127,6 +128,46 @@ type ConvertAutoMlRequest struct { ScoringFunctions string `json:"scoring_functions"` } +type RefConfig struct { + Value string `json:"value"` + ShowValue string `json:"showValue"` + FromSelect bool `json:"fromSelect"` + ActiveTab string `json:"activeTab"` + Id string `json:"id"` + Name string `json:"name"` + Version string `json:"version"` + Path string `json:"path"` + Identifier string `json:"identifier"` + Owner string `json:"owner"` +} + +type OptimizeRequest struct { + CodeConfig CodeConfig `json:"code_config"` + Image RefConfig `json:"image"` + Dataset RefConfig `json:"dataset"` + MainPy string `json:"main_py"` + Name string `json:"name"` + NumSamples int `json:"num_samples"` + PointsToEvaluate []map[string]interface{} `json:"points_to_evaluate"` + Model RefConfig `json:"model"` + SearchAlg string `json:"search_alg"` + Scheduler string `json:"scheduler"` + Metric string `json:"metric"` + Mode string `json:"mode"` + MaxT int `json:"max_t"` + MinSamplesRequired int `json:"min_samples_required"` + Parameters []map[string]interface{} `json:"parameters"` + Resource string `json:"resource"` +} + +var ( + RefTypekeys = []string{ + "dataset", + "model", + "code_config", + } +) + // ConvertAutoMLWorkflow godoc // @Summary 转换工作流 // @Description 前端流水线结构转为argo workflow。 @@ -161,24 +202,302 @@ func (c *ConvertHandler) ConvertAutoMLWorkflow(ctx *gin.Context) { log.Debugf("process pipeline convert request successfully, convert result is %+v", result) } +// ConvertOptimizeWorkflow godoc +// @Summary 转换工作流 +// @Description 前端流水线结构转为argo workflow。 +// @Tags workflow +// @Accept json +// @Produce json +// @Param data body map[string]interface{} true "转换工作流请求参数" +// @Success 200 {object} model.Response "请求成功" +// @Failure 400 {object} model.Response "请求错误" +// @Failure 500 {object} model.Response "内部错误" +// @Router /api/v1/workflow/convertOptimize [post] +func (c *ConvertHandler) ConvertOptimizeWorkflow(ctx *gin.Context) { + var req OptimizeRequest + if err := bindAndValidate(ctx, &req); err != nil { + responseError(ctx, http.StatusBadRequest, err.Error()) + return + } + log.Debugf("optimize req is:%+v\n", req) + dag, err := c.optimizationReqToDag(req) + if err != nil { + responseError(ctx, http.StatusBadRequest, err.Error()) + return + } + + log.Debugf("optimize dag struct is %+v", dag) + result, err := c.service.Convert(ctx, dag) + if err != nil { + log.Errorf("pipline convert yaml err:%v", err) + responseError(ctx, http.StatusBadRequest, err.Error()) + return + } + respondSuccess(ctx, result) + log.Debugf("process optimize pipeline convert request successfully, convert result is %+v", result) +} + func (c *ConvertHandler) convertMap(req map[string]interface{}) map[string]string { result := make(map[string]string, len(req)) - fmt.Printf("map req is:%+v\n", req) + log.Debugf("map req is:%+v\n", req) for k, v := range req { - if k != "dataset" { - result[k] = fmt.Sprintf("%v", v) - } else { + if stringutils.IsContain(RefTypekeys, k) { vauleStr, err := json.Marshal(v) if err != nil { log.Errorf("marshal dataset error:%v", err) return nil } result[k] = string(vauleStr) + } else { + result[k] = fmt.Sprintf("%v", v) } } + log.Debugf("convert map is:%+v", result) return result } +func (c *ConvertHandler) optimizationReqToDag(req OptimizeRequest) (*model.Dag, error) { + frameworkComponent, err := c.buildGitCloneFrameworkComponent() + if err != nil { + return nil, err + } + trainCodeComponent, err := c.buildGitCloneComponent(req) + if err != nil { + return nil, err + } + optimizerComponent, err := c.buildOptimizerComponent(req) + if err != nil { + return nil, err + } + components := []model.Component{ + *frameworkComponent, + *optimizerComponent, + *trainCodeComponent, + } + + lines := []model.Line{ + { + Source: frameworkComponent.TaskID, + Target: optimizerComponent.TaskID, + }, + { + Source: trainCodeComponent.TaskID, + Target: optimizerComponent.TaskID, + }, + } + dag := model.Dag{ + Components: components, + Lines: lines, + } + return &dag, nil +} + +type CodeConfig struct { + Value string `json:"value"` + ShowValue string `json:"showValue"` + FromSelect bool `json:"fromSelect"` + CodePath string `json:"code_path"` + Branch string `json:"branch"` +} + +func (c *ConvertHandler) buildGitCloneComponent(req OptimizeRequest) (*model.Component, error) { + defaultResource := "{\"name\":\"CPU-GPU\",\"value\":{\"detail_type\":\"3060\",\"gpu\":0,\"cpu\":1,\"memory\":\"1GB\"}}" + var resource model.ResourceItem + err := json.Unmarshal([]byte(defaultResource), &resource) + if err != nil { + log.Errorf("unmarshal resource error:%v", err) + return nil, err + } + component := &model.Component{ + CategoryID: parse.GitCloneCategoryID, + ComponentName: parse.GitClone, + ComponentLabel: "代码拉取组件", + Description: "代码拉取组件", + Image: "172.20.32.187/pipeline-component/built-in/git:202409031108", + Resource: resource, + TaskID: parse.GitClone + "-" + stringutils.GetUUID(), + } + + inputParams := make(map[string]model.Param) + inputParams["--code-config"] = model.Param{ + Type: parse.RefType, + ItemType: parse.RefCodeType, + Value: req.CodeConfig.Value, + } + outputParams := make(map[string]model.OutParamItem) + outputParams["--code_output"] = model.OutParamItem{ + Type: parse.StrType, + Path: "/tmp/traincode", + } + component.InParameters = inputParams + component.OutParameters = outputParams + return component, nil +} + +func (c *ConvertHandler) buildGitCloneFrameworkComponent() (*model.Component, error) { + defaultResource := "{\"name\":\"CPU-GPU\",\"value\":{\"detail_type\":\"3060\",\"gpu\":0,\"cpu\":1,\"memory\":\"1GB\"}}" + var resource model.ResourceItem + err := json.Unmarshal([]byte(defaultResource), &resource) + if err != nil { + log.Errorf("unmarshal resource error:%v", err) + return nil, err + } + component := &model.Component{ + CategoryID: parse.GitCloneCategoryID, + ComponentName: parse.GitClone, + ComponentLabel: "代码拉取组件", + Description: "代码拉取组件", + Image: "172.20.32.187/pipeline-component/built-in/git:202409031108", + Resource: resource, + TaskID: parse.GitClone + "-" + stringutils.GetUUID(), + } + + inputParams := make(map[string]model.Param) + inputParams["--code-config"] = model.Param{ + Type: parse.RefType, + ItemType: parse.RefCodeType, + Value: "{\"code_path\":\"https://gitlink.org.cn/chenzhihang/ray_tune.git\",\"branch\":\"master\"}", + } + outputParams := make(map[string]model.OutParamItem) + outputParams["--code_output"] = model.OutParamItem{ + Type: parse.StrType, + Path: "/tmp/frameworkcode", + } + component.InParameters = inputParams + component.OutParameters = outputParams + return component, nil +} + +func (c *ConvertHandler) buildOptimizerComponent(req OptimizeRequest) (*model.Component, error) { + var resource model.ResourceItem + if err := json.Unmarshal([]byte(req.Resource), &resource); err != nil { + log.Errorf("unmarshal resource error:%v", err) + return nil, err + } + components := model.Component{ + CategoryID: parse.AutoMlCategoryID, + ComponentName: parse.HyperparameterOptimization, + ComponentLabel: "超参数优化", + Description: "超参数优化组件", + Image: req.Image.Value, + Resource: resource, + TaskID: parse.HyperparameterOptimization + "-" + stringutils.GetUUID(), + Command: "python /tmp/frameworkcode/main.py", + } + inputParams := make(map[string]model.Param) + inputParams["--name"] = model.Param{ + Type: parse.StrType, + Value: req.Name, + } + inputParams["--main_py"] = model.Param{ + Type: parse.StrType, + Value: req.MainPy, + } + inputParams["--num_samples"] = model.Param{ + Type: parse.StrType, + Value: fmt.Sprintf("%d", req.NumSamples), + } + inputParams["--search_alg"] = model.Param{ + Type: parse.StrType, + Value: req.SearchAlg, + } + inputParams["--scheduler"] = model.Param{ + Type: parse.StrType, + Value: req.Scheduler, + } + inputParams["--metric"] = model.Param{ + Type: parse.StrType, + Value: req.Metric, + } + inputParams["--mode"] = model.Param{ + Type: parse.StrType, + Value: req.Mode, + } + inputParams["--max_t"] = model.Param{ + Type: parse.StrType, + Value: fmt.Sprintf("%d", req.MaxT), + } + inputParams["--min_samples_required"] = model.Param{ + Type: parse.StrType, + Value: fmt.Sprintf("%d", req.MinSamplesRequired), + } + + modelValue, err := json.Marshal(req.Model) + if err != nil { + log.Errorf("marshal model value error:%v", err) + return nil, err + } + inputParams["--model"] = model.Param{ + Type: parse.RefType, + ItemType: parse.RefModelType, + Value: string(modelValue), + } + + datasetValue, err := json.Marshal(req.Dataset) + if err != nil { + log.Errorf("marshal dataset value error:%v", err) + return nil, err + } + inputParams["--dataset"] = model.Param{ + Type: parse.RefType, + ItemType: parse.RefDatasetType, + Value: string(datasetValue), + } + + parametersValue, err := json.Marshal(req.Parameters) + if err != nil { + log.Errorf("marshal parameters value error:%v", err) + return nil, err + } + inputParams["--parameters"] = model.Param{ + Type: parse.StrType, + Value: string(parametersValue), + } + + pointsToEvaluateValue, err := json.Marshal(req.PointsToEvaluate) + if err != nil { + log.Errorf("marshal points to evaluate value error:%v", err) + return nil, err + } + inputParams["--points_to_evaluate"] = model.Param{ + Type: parse.StrType, + Value: string(pointsToEvaluateValue), + } + + inputParams["--cpu"] = model.Param{ + Type: parse.StrType, + Value: fmt.Sprintf("%d", resource.ResourceValue.CPU), + } + // 这里需要将memory从GB转为byte, 首先提取出单位，然后乘以1024 * 1024, memory的格式是2Gi + memory, err := strconv.Atoi(resource.ResourceValue.Memory[:len(resource.ResourceValue.Memory)-2]) + if err != nil { + log.Errorf("convert memory to int error:%v", err) + return nil, err + } + memory = memory * 1024 * 1024 * 1024 + + inputParams["--memory"] = model.Param{ + Type: parse.StrType, + Value: fmt.Sprintf("%d", memory), + } + + if resource.ResourceValue.GPU > 0 { + inputParams["--gpu"] = model.Param{ + Type: parse.StrType, + Value: fmt.Sprintf("%d", resource.ResourceValue.GPU), + } + } + + outputParams := make(map[string]model.OutParamItem) + outputParams["--storage_path"] = model.OutParamItem{ + Type: parse.StrType, + Path: "/tmp/optimizer-output", + } + components.InParameters = inputParams + components.OutParameters = outputParams + return &components, nil +} + func (c *ConvertHandler) autoReqToDag(req map[string]string) (*model.Dag, error) { defaultResource := "{\"name\":\"CPU-GPU\",\"value\":{\"detail_type\":\"3060\",\"gpu\":0,\"cpu\":4,\"memory\":\"8GB\"}}" var resource model.ResourceItem diff --git a/internal/router/api_router.go b/internal/router/api_router.go index fdf55f5..ecd6ce1 100644 --- a/internal/router/api_router.go +++ b/internal/router/api_router.go @@ -34,6 +34,7 @@ func SetupApiRouters(r *gin.Engine) { v1 := r.Group("/api/v1") v1.POST("workflow/convert", connvertHandler.ConvertWorkflow) v1.POST("workflow/convertAutoML", connvertHandler.ConvertAutoMLWorkflow) + v1.POST("workflow/convertOptimize", connvertHandler.ConvertOptimizeWorkflow) v1.POST("workflow/copy", connvertHandler.CopyWorkflow) v1.POST("workflow/run", runHandler.RunWorkflow) v1.POST("workflow/getWorkflow", runHandler.GetWorkflow) diff --git a/internal/service/parse/parse.go b/internal/service/parse/parse.go index f2ae25f..76e20f9 100644 --- a/internal/service/parse/parse.go +++ b/internal/service/parse/parse.go @@ -29,7 +29,9 @@ const ( GeneralDataProcess = "general-data-process" GitClone = "git-clone" AutoMlCategoryID = 8 + GitCloneCategoryID = 1 AutoML = "auto-ml" + HyperparameterOptimization = "auto-hpo" ExperimentRemoteRepo = "EXPERIMENT_REMOTE_REPO" ExperimentRunID = "EXPERIMENT_RUN_ID" ExperimentRunName = "EXPERIMENT_RUN_NAME" diff --git a/pipeline-convert.yaml b/pipeline-convert.yaml index afae67a..36ceff3 100644 --- a/pipeline-convert.yaml +++ b/pipeline-convert.yaml @@ -45,7 +45,19 @@ spec: protocol: TCP nodePort: 31000 - +--- +kind: PersistentVolumeClaim +apiVersion: v1 +metadata: + name: pipeline-convert-log-pvc-nfs + namespace: argo +spec: + storageClassName: nfs-client + accessModes: + - ReadWriteMany + resources: + requests: + storage: 5Gi