超参寻优workflow转换

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
+scp pipeline-convert.yaml root@172.20.32.197:/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"
+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"

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

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" {
+		if stringutils.IsContain(RefTypekeys, k) {
-			result[k] = fmt.Sprintf("%v", v)
-		} else {
 			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

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)

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"

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